opensearch-protobufs

import grpc import os import base64 from opensearch.protobufs.schemas import SearchRequest, Query, MatchQuery from opensearch.protobufs.services import SearchServiceStub # Configure your OpenSearch gRPC endpoint # In a real application, you'd replace this with your actual OpenSearch gRPC host and port OPENSEARCH_GRPC_TARGET = os.environ.get('OPENSEARCH_GRPC_TARGET', 'localhost:9200') # Example def run_search_query(): try: # Establish an insecure gRPC channel (use secure channels for production!) with grpc.insecure_channel(OPENSEARCH_GRPC_TARGET) as channel: stub = SearchServiceStub(channel) # Create a simple match query match_query = MatchQuery(field='title', query='OpenSearch') query = Query(match=match_query) # Create the search request search_request = SearchRequest( query=query, size=5 ) print(f"Sending SearchRequest to {OPENSEARCH_GRPC_TARGET}:") print(search_request) # Make the gRPC call response = stub.Search(search_request) print("\nSearch Response:") for hit in response.hits.hits: # _source is returned as bytes, decode if it's JSON-like source_data = hit.source.value.decode('utf-8') if hit.source.value else '{}' print(f" ID: {hit.id}, Score: {hit.score}, Source: {source_data}") except grpc.RpcError as e: print(f"gRPC Error: {e.code()} - {e.details()}") except Exception as e: print(f"An unexpected error occurred: {e}") if __name__ == '__main__': print("Note: This quickstart requires a running OpenSearch instance with gRPC enabled.") print(f"Attempting to connect to gRPC target: {OPENSEARCH_GRPC_TARGET}\n") run_search_query()