azure-search

import os from azure.core.credentials import AzureKeyCredential from azure.search.documents import SearchClient from azure.search.documents.indexes import SearchIndexClient from azure.search.documents.models import SearchableField, SearchFieldDataType, SimpleField, SearchIndex # Replace with your Azure AI Search service endpoint and admin key service_endpoint = os.environ.get("SEARCH_SERVICE_ENDPOINT", "https://YOUR_SEARCH_SERVICE_NAME.search.windows.net") admin_key = os.environ.get("SEARCH_ADMIN_KEY", "YOUR_ADMIN_KEY") index_name = "test-index" if not service_endpoint or not admin_key: print("Please set the SEARCH_SERVICE_ENDPOINT and SEARCH_ADMIN_KEY environment variables.") exit(1) # Create a SearchIndexClient credential = AzureKeyCredential(admin_key) index_client = SearchIndexClient(endpoint=service_endpoint, credential=credential) # Define a search index fields = [ SimpleField(name="id", type=SearchFieldDataType.String, key=True), SearchableField(name="title", type=SearchFieldDataType.String, searchable=True), SearchableField(name="description", type=SearchFieldDataType.String, searchable=True), ] index = SearchIndex(name=index_name, fields=fields) # Create the index try: print(f"Creating index '{index_name}'...") index_client.create_index(index) print("Index created.") # Get a SearchClient to add and search documents search_client = SearchClient(endpoint=service_endpoint, index_name=index_name, credential=credential) # Add documents documents = [ {"id": "1", "title": "Azure AI Search Document 1", "description": "This is the first document."}, {"id": "2", "title": "Azure AI Search Document 2", "description": "This is the second document."} ] print(f"Uploading {len(documents)} documents...") search_client.upload_documents(documents) print("Documents uploaded.") # Search documents print("Searching for 'first'...") results = search_client.search(search_text="first") for result in results: print(f"Found document: {result['title']}") finally: # Clean up (delete index) print(f"Deleting index '{index_name}'...") index_client.delete_index(index_name) print("Index deleted.")