OpenSearch Protobufs Python Client

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()