From f1d1d59e7ec016dcb9eec8dd5241c977c6cdad6e Mon Sep 17 00:00:00 2001 From: noel-improv Date: Tue, 14 Jul 2026 15:46:52 -0600 Subject: [PATCH 01/10] feat(lexical-graph): add opensearch:// local vector store support Add an opensearch:// connection string for self-managed (non-AWS) OpenSearch instances, alongside the existing aoss:// one. The local path skips AWS SigV4 entirely: it authenticates with HTTP basic auth from OPENSEARCH_USERNAME/OPENSEARCH_PASSWORD (env vars or GraphRAGConfig), or with no auth if neither is set. index_exists() now threads client_kwargs through to the OpenSearch client, so callers can override use_ssl/verify_certs for a plain-HTTP or self-signed local endpoint. The existing Classic knn_vector mapping is reused unchanged. --- .../graphrag_toolkit/lexical_graph/config.py | 27 ++ .../vector/opensearch_vector_index_factory.py | 15 +- .../vector/opensearch_vector_indexes.py | 102 +++++--- .../storage/vector/test_opensearch_clients.py | 243 ++++++++++++++++++ .../vector/test_vector_index_factories.py | 72 +++++- lexical-graph/tests/unit/test_config.py | 60 +++++ 6 files changed, 480 insertions(+), 39 deletions(-) diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py index 0a2612fc2..5d5999c7e 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py @@ -291,6 +291,8 @@ class _GraphRAGConfig: _enable_cache: Optional[bool] = None _metadata_datetime_suffixes: Optional[List[str]] = None _opensearch_engine: Optional[str] = None + _opensearch_username: Optional[str] = None + _opensearch_password: Optional[str] = None _enable_versioning = None _chunk_external_properties: Optional[Dict[str, str]] = None _local_output_dir: Optional[str] = None @@ -1172,6 +1174,31 @@ def opensearch_engine(self) -> str: def opensearch_engine(self, opensearch_engine: str) -> None: self._opensearch_engine = opensearch_engine + @property + def opensearch_username(self) -> Optional[str]: + """Basic-auth username for a local (non-AWS) OpenSearch endpoint. Unset (None) + means no basic auth is applied -- used with a local OpenSearch instance that has + no security plugin configured.""" + if self._opensearch_username is None: + self._opensearch_username = os.environ.get('OPENSEARCH_USERNAME') + return self._opensearch_username + + @opensearch_username.setter + def opensearch_username(self, opensearch_username: Optional[str]) -> None: + self._opensearch_username = opensearch_username + + @property + def opensearch_password(self) -> Optional[str]: + """Basic-auth password for a local (non-AWS) OpenSearch endpoint. See + opensearch_username.""" + if self._opensearch_password is None: + self._opensearch_password = os.environ.get('OPENSEARCH_PASSWORD') + return self._opensearch_password + + @opensearch_password.setter + def opensearch_password(self, opensearch_password: Optional[str]) -> None: + self._opensearch_password = opensearch_password + @property def enable_versioning(self) -> bool: if self._enable_versioning is None: diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index bf620d440..6ac4186b7 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -10,6 +10,7 @@ OPENSEARCH_SERVERLESS = 'aoss://' OPENSEARCH_SERVERLESS_DNS = 'aoss.amazonaws.com' +OPENSEARCH_LOCAL = 'opensearch://' class OpenSearchVectorIndexFactory(VectorIndexFactoryMethod): """Factory class for creating OpenSearch vector indexes. @@ -43,19 +44,25 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> names and endpoint configuration, or None if no suitable endpoint is found. """ endpoint = None + is_local = False if vector_index_info.startswith(OPENSEARCH_SERVERLESS): endpoint = vector_index_info[len(OPENSEARCH_SERVERLESS):] if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' + elif vector_index_info.startswith(OPENSEARCH_LOCAL): + endpoint = vector_index_info[len(OPENSEARCH_LOCAL):] + if not endpoint.startswith('https://') and not endpoint.startswith('http://'): + endpoint = f'https://{endpoint}' + is_local = True elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info if endpoint: try: from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes import OpenSearchIndex - logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}]') - return [OpenSearchIndex.for_index(index_name, endpoint, **kwargs) for index_name in index_names] + logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_local: {is_local}]') + return [OpenSearchIndex.for_index(index_name, endpoint, is_local=is_local, **kwargs) for index_name in index_names] except ImportError as e: raise e - + else: - return None \ No newline at end of file + return None diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index c39e98606..820019251 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -115,20 +115,33 @@ class DummyAuth: DEFAULT_POOL_MAXSIZE = 32 -def create_os_client(endpoint, **kwargs): +def _local_http_auth(): + """HTTP basic-auth tuple for a local OpenSearch endpoint from + GraphRAGConfig.opensearch_username/opensearch_password, or None if either is unset + (a local instance with no security plugin configured).""" + username = GraphRAGConfig.opensearch_username + password = GraphRAGConfig.opensearch_password + if username and password: + return (username, password) + if username or password: + logger.warning('Only one of opensearch_username/opensearch_password is set; connecting without basic auth') + return None + +def create_os_client(endpoint, is_local=False, **kwargs): """ - Creates an OpenSearch client configured to use AWS Signature Version 4 - authentication. + Creates an OpenSearch client. - The function utilizes AWS credentials from a pre-configured session and region - information. It signs requests using Urllib3-based AWS Signature V4 - authentication. The client is initialized with specific connection properties - like SSL usage, certificate verification, timeout, retry settings, and any - additional keyword arguments provided. + By default the client is configured to use AWS Signature Version 4 authentication, + signed with credentials from a pre-configured session and region. When is_local is + True, this instead connects to a self-managed OpenSearch instance: HTTP basic auth + from GraphRAGConfig.opensearch_username/opensearch_password if both are set, + otherwise no auth. The AWS session and region are never touched on the local path. Args: endpoint: str The OpenSearch endpoint URL to connect to. + is_local: bool + True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. **kwargs: Any Additional keyword arguments passed to the OpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. @@ -138,12 +151,15 @@ def create_os_client(endpoint, **kwargs): A configured client instance for interacting with OpenSearch. """ - session = GraphRAGConfig.session - region = GraphRAGConfig.aws_region - credentials = session.get_credentials() - service = 'aoss' + if is_local: + auth = _local_http_auth() + else: + session = GraphRAGConfig.session + region = GraphRAGConfig.aws_region + credentials = session.get_credentials() + service = 'aoss' - auth = Urllib3AWSV4SignerAuth(credentials, region, service) + auth = Urllib3AWSV4SignerAuth(credentials, region, service) defaults = { 'use_ssl': True, @@ -161,28 +177,34 @@ def create_os_client(endpoint, **kwargs): **{**defaults, **kwargs}, ) -def create_os_async_client(endpoint, **kwargs): +def create_os_async_client(endpoint, is_local=False, **kwargs): """ Creates an asynchronous OpenSearch client. - This function configures and returns an instance of the AsyncOpenSearch client, - configured for secure communication with the specified endpoint. The client - leverages AWS Signature Version 4 for authentication. + By default this authenticates with AWS Signature Version 4. When is_local is True, + it connects to a self-managed OpenSearch instance instead: HTTP basic auth from + GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no + auth. The AWS session and region are never touched on the local path. Args: endpoint: The URL of the OpenSearch cluster endpoint. + is_local: bool + True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. **kwargs: Optional parameters for customizing the AsyncOpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. Returns: AsyncOpenSearch: An instantiated asynchronous OpenSearch client. """ - session = GraphRAGConfig.session - region = GraphRAGConfig.aws_region - credentials = session.get_credentials() - service = 'aoss' + if is_local: + auth = _local_http_auth() + else: + session = GraphRAGConfig.session + region = GraphRAGConfig.aws_region + credentials = session.get_credentials() + service = 'aoss' - auth = AWSV4SignerAsyncAuth(credentials, region, service) + auth = AWSV4SignerAsyncAuth(credentials, region, service) defaults = { 'use_ssl': True, @@ -228,7 +250,7 @@ def index_is_available(client, index_name): -def index_exists(endpoint, index_name, dimensions, writeable) -> bool: +def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, client_kwargs=None) -> bool: """ Checks if an OpenSearch index exists, and optionally creates it if it does not exist. @@ -242,11 +264,16 @@ def index_exists(endpoint, index_name, dimensions, writeable) -> bool: index_name: The name of the index to check for existence. dimensions: The number of dimensions for the knn_vector in the index. writeable: Flag indicating whether to create the index if it does not exist. + is_local: True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. + client_kwargs: Optional dict of keyword arguments passed through to the + OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP local + endpoint). ``pool_maxsize`` defaults to 1 unless overridden here. Returns: bool: True if the index exists (or is created successfully), False otherwise. """ - client = create_os_client(endpoint, pool_maxsize=1) + client_kwargs = {'pool_maxsize': 1, **(client_kwargs or {})} + client = create_os_client(endpoint, is_local=is_local, **client_kwargs) embedding_field = 'embedding' @@ -314,7 +341,7 @@ def index_exists(endpoint, index_name, dimensions, writeable) -> bool: return index_exists -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None, is_local=False): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -332,6 +359,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode ``{"pool_maxsize": 10, "timeout": 60}``). Defaults that are already set on the clients (``use_ssl``, ``verify_certs``, ``connection_class``, ``timeout``, ``max_retries``, ``retry_on_timeout``) can be overridden here. + is_local: True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4 + and marks the underlying client as non-AOSS (refresh() is called after + bulk ingest, matching a self-managed cluster's behavior). Returns: OpensearchVectorClient: A configured OpenSearch vector client instance. @@ -345,7 +375,7 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode client_kwargs = client_kwargs or {} - logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}]') + logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_local={is_local}]') client = None retry_count = 0 @@ -357,9 +387,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode dimensions, embedding_field=embedding_field, text_field=text_field, - os_client=create_os_client(endpoint, **client_kwargs), - os_async_client=create_os_async_client(endpoint, **client_kwargs), - http_auth=DummyAuth(service='aoss') + os_client=create_os_client(endpoint, is_local=is_local, **client_kwargs), + os_async_client=create_os_async_client(endpoint, is_local=is_local, **client_kwargs), + http_auth=DummyAuth(service='local') if is_local else DummyAuth(service='aoss') ) start = time.time() @@ -478,7 +508,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None, is_local=False): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. @@ -498,6 +528,8 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw client_kwargs (Optional[Dict[str, Any]]): Optional dict of keyword arguments forwarded to the underlying synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). + is_local (bool): True for a self-managed (non-AWS) OpenSearch endpoint; + skips AWS SigV4 signing in favor of HTTP basic auth or no auth. Returns: OpenSearchIndex: An instance of OpenSearchIndex initialized with the @@ -506,7 +538,7 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw embed_model = embed_model or GraphRAGConfig.embed_model dimensions = dimensions or GraphRAGConfig.embed_dimensions - return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, client_kwargs=client_kwargs) + return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, client_kwargs=client_kwargs, is_local=is_local) endpoint:str index_name:str @@ -514,6 +546,7 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw embed_model:EmbeddingType model_config = ConfigDict(arbitrary_types_allowed=True) client_kwargs:Optional[Dict[str, Any]]=None + is_local:bool=False _client: OpensearchVectorClient = PrivateAttr(default=None) @@ -556,20 +589,21 @@ def client(self) -> OpensearchVectorClient: self._client = None if not self._client: - if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable): + if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_local=self.is_local, client_kwargs=self.client_kwargs): self._client = create_opensearch_vector_client( self.endpoint, self.underlying_index_name(), self.dimensions, self.embed_model, client_kwargs=self.client_kwargs, + is_local=self.is_local, ) else: self._client = DummyOpensearchVectorClient() return self._client - + def index_exists(self): - return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable) + return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_local=self.is_local, client_kwargs=self.client_kwargs) def _clean_id(self, s): """ diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index a46142aa0..510a094ce 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -8,6 +8,7 @@ - create_os_async_client (returns AsyncOpenSearch instance) - create_opensearch_vector_client (success, NotFoundError retry) - OpenSearchIndex.client property (index exists, index does not exist) + - create_os_client/create_os_async_client is_local path (basic auth, no auth, no SigV4) """ import sys @@ -235,3 +236,245 @@ def test_client_creates_real_client_when_index_exists(self): with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client): result = index.client assert result is mock_vector_client + + +# --------------------------------------------------------------------------- +# create_os_client / create_os_async_client: is_local path +# --------------------------------------------------------------------------- + +class TestCreateOsClientLocal: + + def test_local_with_credentials_uses_basic_auth(self): + mock_os_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: + result = ovi.create_os_client("https://localhost:9200", is_local=True) + assert result is mock_os_instance + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] == ("admin", "secret") + + def test_local_without_credentials_uses_no_auth(self): + mock_os_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: + ovi.create_os_client("https://localhost:9200", is_local=True) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + + def test_local_with_only_username_warns_and_uses_no_auth(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_local=True) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + assert any("Only one of" in r.message for r in caplog.records) + + def test_local_with_only_password_warns_and_uses_no_auth(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch") as mock_os: + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_local=True) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + assert any("Only one of" in r.message for r in caplog.records) + + def test_local_never_touches_aws_session_or_sigv4(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch"): + with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: + ovi.create_os_client("https://localhost:9200", is_local=True) + mock_sigv4.assert_not_called() + mock_cfg.session.assert_not_called() + + def test_default_is_local_false_preserves_sigv4(self): + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "OpenSearch"): + with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: + ovi.create_os_client("https://my-endpoint.aoss.amazonaws.com") + mock_sigv4.assert_called_once() + + +class TestCreateOsAsyncClientLocal: + + def test_local_with_credentials_uses_basic_auth(self): + mock_async_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "AsyncOpenSearch", return_value=mock_async_instance) as mock_async: + result = ovi.create_os_async_client("https://localhost:9200", is_local=True) + assert result is mock_async_instance + _, kwargs = mock_async.call_args + assert kwargs["http_auth"] == ("admin", "secret") + + def test_local_without_credentials_uses_no_auth(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + ovi.create_os_async_client("https://localhost:9200", is_local=True) + _, kwargs = mock_async.call_args + assert kwargs["http_auth"] is None + + def test_local_never_touches_aws_session_or_sigv4(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch"): + with patch.object(ovi, "AWSV4SignerAsyncAuth") as mock_sigv4: + ovi.create_os_async_client("https://localhost:9200", is_local=True) + mock_sigv4.assert_not_called() + mock_cfg.session.assert_not_called() + + +class TestCreateOpensearchVectorClientLocal: + + def test_is_local_threaded_to_both_clients(self): + with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): + with patch.object(ovi, "create_os_client", return_value=MagicMock()) as mock_sync: + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()) as mock_async: + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://localhost:9200", "my-index", 1024, MagicMock(), is_local=True + ) + assert mock_sync.call_args.kwargs.get("is_local") is True + assert mock_async.call_args.kwargs.get("is_local") is True + + def test_is_local_sets_non_aoss_dummy_auth_service(self): + captured = {} + + def fake_vector_client(*args, **kwargs): + captured["http_auth"] = kwargs.get("http_auth") + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=fake_vector_client): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://localhost:9200", "my-index", 1024, MagicMock(), is_local=True + ) + assert captured["http_auth"].service != "aoss" + + def test_default_is_local_false_keeps_aoss_dummy_auth_service(self): + captured = {} + + def fake_vector_client(*args, **kwargs): + captured["http_auth"] = kwargs.get("http_auth") + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=fake_vector_client): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock() + ) + assert captured["http_auth"].service == "aoss" + + +class TestOpenSearchIndexClientPropertyLocal: + + def _make_local_index(self): + from graphrag_toolkit.lexical_graph.tenant_id import TenantId + return ovi.OpenSearchIndex.model_construct( + index_name="chunk", + endpoint="https://localhost:9200", + dimensions=1024, + embed_model=MagicMock(), + tenant_id=TenantId(), + writeable=False, + is_local=True, + _client=None, + ) + + def test_client_property_threads_is_local_to_index_exists(self): + index = self._make_local_index() + + with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: + index.client + assert mock_index_exists.call_args.kwargs.get("is_local") is True + + def test_client_property_threads_is_local_to_vector_client(self): + index = self._make_local_index() + mock_vector_client = MagicMock() + + with patch.object(ovi, "index_exists", return_value=True): + with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client) as mock_create: + index.client + assert mock_create.call_args.kwargs.get("is_local") is True + + def test_client_property_threads_client_kwargs_to_index_exists(self): + from graphrag_toolkit.lexical_graph.tenant_id import TenantId + index = ovi.OpenSearchIndex.model_construct( + index_name="chunk", + endpoint="http://localhost:9200", + dimensions=1024, + embed_model=MagicMock(), + tenant_id=TenantId(), + writeable=False, + is_local=True, + client_kwargs={"use_ssl": False, "verify_certs": False}, + _client=None, + ) + + with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: + index.client + assert mock_index_exists.call_args.kwargs.get("client_kwargs") == {"use_ssl": False, "verify_certs": False} + + +# --------------------------------------------------------------------------- +# index_exists: client_kwargs passthrough (plain-HTTP / self-signed local endpoints) +# --------------------------------------------------------------------------- + +class TestIndexExistsClientKwargs: + + def _mock_client(self): + client = MagicMock() + client.indices.exists.return_value = True + return client + + def test_default_client_kwargs_none_still_sets_pool_maxsize(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists("https://endpoint", "my-index", 1024, writeable=False) + _, kwargs = mock_create.call_args + assert kwargs["pool_maxsize"] == 1 + + def test_client_kwargs_forwarded_to_create_os_client(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists( + "http://localhost:9200", "my-index", 1024, writeable=False, is_local=True, + client_kwargs={"use_ssl": False, "verify_certs": False}, + ) + _, kwargs = mock_create.call_args + assert kwargs["use_ssl"] is False + assert kwargs["verify_certs"] is False + assert kwargs["is_local"] is True + + def test_client_kwargs_can_override_default_pool_maxsize(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists( + "https://endpoint", "my-index", 1024, writeable=False, + client_kwargs={"pool_maxsize": 10}, + ) + _, kwargs = mock_create.call_args + assert kwargs["pool_maxsize"] == 10 diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index 4d0b78223..9e3fe27d7 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -4,7 +4,7 @@ """Tests for vector index factory try_create methods. Covers: - - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, no match) + - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, opensearch://, no match) - PGVectorIndexFactory.try_create (postgres://, postgresql://, no match) - S3VectorIndexFactory.try_create (s3vectors://, no match) """ @@ -67,6 +67,76 @@ def test_try_create_https_aoss_dns_returns_indexes(self): ) assert result is not None + def test_try_create_opensearch_prefix_returns_indexes(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_index = MagicMock() + mock_cls = MagicMock() + mock_cls.for_index.return_value = mock_index + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + result = factory.try_create( + ["chunk", "statement"], + "opensearch://localhost:9200" + ) + assert result is not None + assert len(result) == 2 + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_local") is True + + def test_try_create_opensearch_prefix_defaults_to_https(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "opensearch://localhost:9200") + args, _ = mock_cls.for_index.call_args + assert args[1] == "https://localhost:9200" + + def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "opensearch://http://localhost:9200") + args, _ = mock_cls.for_index.call_args + assert args[1] == "http://localhost:9200" + + def test_try_create_aoss_prefix_does_not_set_is_local(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "aoss://https://abc.us-east-1.aoss.amazonaws.com") + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_local") is not True + # --------------------------------------------------------------------------- # PGVectorIndexFactory.try_create diff --git a/lexical-graph/tests/unit/test_config.py b/lexical-graph/tests/unit/test_config.py index e2846b02a..29b3efb11 100644 --- a/lexical-graph/tests/unit/test_config.py +++ b/lexical-graph/tests/unit/test_config.py @@ -204,3 +204,63 @@ def test_build_batch_size_from_env(self): assert GraphRAGConfig.build_batch_size == 7 finally: GraphRAGConfig._build_batch_size = original + + def test_opensearch_username_defaults_none(self): + """Verify opensearch_username defaults to None when unset.""" + original = GraphRAGConfig._opensearch_username + try: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop('OPENSEARCH_USERNAME', None) + GraphRAGConfig._opensearch_username = None + assert GraphRAGConfig.opensearch_username is None + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_username_from_env(self): + """Verify opensearch_username reads from OPENSEARCH_USERNAME env var.""" + original = GraphRAGConfig._opensearch_username + try: + with patch.dict(os.environ, {'OPENSEARCH_USERNAME': 'admin'}): + GraphRAGConfig._opensearch_username = None + assert GraphRAGConfig.opensearch_username == 'admin' + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_username_setter(self): + """Verify opensearch_username can be set directly.""" + original = GraphRAGConfig._opensearch_username + try: + GraphRAGConfig.opensearch_username = 'custom-user' + assert GraphRAGConfig.opensearch_username == 'custom-user' + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_password_defaults_none(self): + """Verify opensearch_password defaults to None when unset.""" + original = GraphRAGConfig._opensearch_password + try: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop('OPENSEARCH_PASSWORD', None) + GraphRAGConfig._opensearch_password = None + assert GraphRAGConfig.opensearch_password is None + finally: + GraphRAGConfig._opensearch_password = original + + def test_opensearch_password_from_env(self): + """Verify opensearch_password reads from OPENSEARCH_PASSWORD env var.""" + original = GraphRAGConfig._opensearch_password + try: + with patch.dict(os.environ, {'OPENSEARCH_PASSWORD': 'secret'}): + GraphRAGConfig._opensearch_password = None + assert GraphRAGConfig.opensearch_password == 'secret' + finally: + GraphRAGConfig._opensearch_password = original + + def test_opensearch_password_setter(self): + """Verify opensearch_password can be set directly.""" + original = GraphRAGConfig._opensearch_password + try: + GraphRAGConfig.opensearch_password = 'custom-pass' + assert GraphRAGConfig.opensearch_password == 'custom-pass' + finally: + GraphRAGConfig._opensearch_password = original From 3e06dc0cad43d335e56759cd4de706819ea64810 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Tue, 14 Jul 2026 15:47:03 -0600 Subject: [PATCH 02/10] docs(lexical-graph): document the opensearch:// local vector store --- .../vector-store-opensearch-serverless.mdx | 19 +++++++++++++++++++ lexical-graph/README.md | 5 ++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index b18a08391..231401fd6 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -9,6 +9,7 @@ title: OpenSearch Serverless Vector Store - [Creating an OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) - [Amazon OpenSearch Serverless and custom document IDs](#amazon-opensearch-serverless-and-custom-document-ids) - [Verify and repair an Amazon OpenSearch Serverless vector store](#verify-and-repair-an-amazon-opensearch-serverless-vector-store) + - [Self-managed OpenSearch](#self-managed-opensearch) ### Overview @@ -130,3 +131,21 @@ Field descriptions: | `num_docs` | Number of documents in a specific tenant vector index | | `num_deleted` | Number of documents deleted from a specific tenant vector index (indicative number only if `dry_run` is `true`) | | `num_unindexed` | Number of nodes that have not been indexed in a specific tenant vector index | + +### Self-managed OpenSearch + +You can also use a self-managed (non-AWS) OpenSearch instance — for example, a local Docker or Finch container — as a vector store. Supply a connection string that begins `opensearch://`, followed by the endpoint: + +```python +from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory + +opensearch_connection_info = 'opensearch://http://localhost:9200' + +with VectorStoreFactory.for_vector_store( + opensearch_connection_info, + index_names=['chunk'] +) as vector_store: + ... +``` + +Unlike `aoss://`, a self-managed endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. If you omit the scheme from the endpoint (e.g. `opensearch://localhost:9200`), it defaults to `https://`. diff --git a/lexical-graph/README.md b/lexical-graph/README.md index b80382a15..5fa70ed44 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -41,7 +41,7 @@ If you're running on AWS, you must run your application in an AWS region contain You will need to install additional dependencies for specific graph and vector store backends: -#### Amazon OpenSearch Serverless +#### OpenSearch (Amazon OpenSearch Serverless or self-managed) ```bash $ pip install opensearch-py llama-index-vector-stores-opensearch @@ -69,11 +69,14 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | Neptune Database (graph) | `neptune-db://` or any hostname ending `.neptune.amazonaws.com` | | Neo4j (graph) | `bolt://`, `bolt+ssc://`, `bolt+s://`, `neo4j://`, `neo4j+ssc://`, or `neo4j+s://` URLs | | OpenSearch Serverless (vector) | `aoss://` | +| OpenSearch, self-managed (vector) | `opensearch://` | | Neptune Analytics (vector) | `neptune-graph://` | | pgvector (vector) | constructed via `PGVectorIndexFactory` | | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | | Dummy / no-op | `None` or any unrecognised string — falls back to `DummyGraphStore` / `DummyVectorIndex` | +A self-managed OpenSearch endpoint (`opensearch://`) connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://`, it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. + ## Example of use ### Indexing From 467d353943693a5966b3ef789bb65ec31c12304f Mon Sep 17 00:00:00 2001 From: noel-improv Date: Wed, 15 Jul 2026 16:58:42 -0600 Subject: [PATCH 03/10] fix(lexical-graph): use OpenSearch/AOSS terminology, keep client_kwargs last Docs and docstrings previously called the new non-AOSS path "self-managed" or "local" OpenSearch, which understates it -- OpenSearch runs anywhere (local, EC2, on-prem), and "local" only covers one case. Switch to plain "OpenSearch" for the open source path and "Amazon OpenSearch Serverless (AOSS)" for the managed one, matching the existing env var naming convention. Also make client_kwargs consistently the last parameter across create_opensearch_vector_client, OpenSearchIndex.for_index, and the OpenSearchIndex model fields -- it was sometimes before is_local, sometimes after. --- .../vector-store-opensearch-serverless.mdx | 8 +-- lexical-graph/README.md | 8 +-- .../graphrag_toolkit/lexical_graph/config.py | 10 ++-- .../vector/opensearch_vector_indexes.py | 50 +++++++++++-------- 4 files changed, 41 insertions(+), 35 deletions(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index 231401fd6..90b3f6080 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -9,7 +9,7 @@ title: OpenSearch Serverless Vector Store - [Creating an OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) - [Amazon OpenSearch Serverless and custom document IDs](#amazon-opensearch-serverless-and-custom-document-ids) - [Verify and repair an Amazon OpenSearch Serverless vector store](#verify-and-repair-an-amazon-opensearch-serverless-vector-store) - - [Self-managed OpenSearch](#self-managed-opensearch) + - [OpenSearch](#opensearch) ### Overview @@ -132,9 +132,9 @@ Field descriptions: | `num_deleted` | Number of documents deleted from a specific tenant vector index (indicative number only if `dry_run` is `true`) | | `num_unindexed` | Number of nodes that have not been indexed in a specific tenant vector index | -### Self-managed OpenSearch +### OpenSearch -You can also use a self-managed (non-AWS) OpenSearch instance — for example, a local Docker or Finch container — as a vector store. Supply a connection string that begins `opensearch://`, followed by the endpoint: +You can also use OpenSearch directly as a vector store, instead of Amazon OpenSearch Serverless — running locally in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. Supply a connection string that begins `opensearch://`, followed by the endpoint: ```python from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory @@ -148,4 +148,4 @@ with VectorStoreFactory.for_vector_store( ... ``` -Unlike `aoss://`, a self-managed endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. If you omit the scheme from the endpoint (e.g. `opensearch://localhost:9200`), it defaults to `https://`. +Unlike `aoss://` (Amazon OpenSearch Serverless), an `opensearch://` endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. If you omit the scheme from the endpoint (e.g. `opensearch://localhost:9200`), it defaults to `https://`. diff --git a/lexical-graph/README.md b/lexical-graph/README.md index 5fa70ed44..f4eb765b0 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -41,7 +41,7 @@ If you're running on AWS, you must run your application in an AWS region contain You will need to install additional dependencies for specific graph and vector store backends: -#### OpenSearch (Amazon OpenSearch Serverless or self-managed) +#### OpenSearch and Amazon OpenSearch Serverless (AOSS) ```bash $ pip install opensearch-py llama-index-vector-stores-opensearch @@ -68,14 +68,14 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | Neptune Analytics (graph) | `neptune-graph://` | | Neptune Database (graph) | `neptune-db://` or any hostname ending `.neptune.amazonaws.com` | | Neo4j (graph) | `bolt://`, `bolt+ssc://`, `bolt+s://`, `neo4j://`, `neo4j+ssc://`, or `neo4j+s://` URLs | -| OpenSearch Serverless (vector) | `aoss://` | -| OpenSearch, self-managed (vector) | `opensearch://` | +| Amazon OpenSearch Serverless / AOSS (vector) | `aoss://` | +| OpenSearch (vector) | `opensearch://` | | Neptune Analytics (vector) | `neptune-graph://` | | pgvector (vector) | constructed via `PGVectorIndexFactory` | | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | | Dummy / no-op | `None` or any unrecognised string — falls back to `DummyGraphStore` / `DummyVectorIndex` | -A self-managed OpenSearch endpoint (`opensearch://`) connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://`, it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. +An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://` (Amazon OpenSearch Serverless), it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. ## Example of use diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py index 5d5999c7e..10ae5177b 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py @@ -1176,9 +1176,9 @@ def opensearch_engine(self, opensearch_engine: str) -> None: @property def opensearch_username(self) -> Optional[str]: - """Basic-auth username for a local (non-AWS) OpenSearch endpoint. Unset (None) - means no basic auth is applied -- used with a local OpenSearch instance that has - no security plugin configured.""" + """Basic-auth username for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS). Unset (None) means no basic auth is applied -- used with an + OpenSearch instance that has no security plugin configured.""" if self._opensearch_username is None: self._opensearch_username = os.environ.get('OPENSEARCH_USERNAME') return self._opensearch_username @@ -1189,8 +1189,8 @@ def opensearch_username(self, opensearch_username: Optional[str]) -> None: @property def opensearch_password(self) -> Optional[str]: - """Basic-auth password for a local (non-AWS) OpenSearch endpoint. See - opensearch_username.""" + """Basic-auth password for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS). See opensearch_username.""" if self._opensearch_password is None: self._opensearch_password = os.environ.get('OPENSEARCH_PASSWORD') return self._opensearch_password diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index 820019251..adf04d530 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -116,9 +116,9 @@ class DummyAuth: DEFAULT_POOL_MAXSIZE = 32 def _local_http_auth(): - """HTTP basic-auth tuple for a local OpenSearch endpoint from - GraphRAGConfig.opensearch_username/opensearch_password, or None if either is unset - (a local instance with no security plugin configured).""" + """HTTP basic-auth tuple for an OpenSearch endpoint (not Amazon OpenSearch + Serverless) from GraphRAGConfig.opensearch_username/opensearch_password, or None if + either is unset (an instance with no security plugin configured).""" username = GraphRAGConfig.opensearch_username password = GraphRAGConfig.opensearch_password if username and password: @@ -133,15 +133,17 @@ def create_os_client(endpoint, is_local=False, **kwargs): By default the client is configured to use AWS Signature Version 4 authentication, signed with credentials from a pre-configured session and region. When is_local is - True, this instead connects to a self-managed OpenSearch instance: HTTP basic auth - from GraphRAGConfig.opensearch_username/opensearch_password if both are set, - otherwise no auth. The AWS session and region are never touched on the local path. + True, this instead connects to an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS): HTTP basic auth from + GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise + no auth. The AWS session and region are never touched on this path. Args: endpoint: str The OpenSearch endpoint URL to connect to. is_local: bool - True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. + True for an OpenSearch endpoint that isn't Amazon OpenSearch Serverless + (AOSS); skips SigV4. **kwargs: Any Additional keyword arguments passed to the OpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. @@ -182,14 +184,16 @@ def create_os_async_client(endpoint, is_local=False, **kwargs): Creates an asynchronous OpenSearch client. By default this authenticates with AWS Signature Version 4. When is_local is True, - it connects to a self-managed OpenSearch instance instead: HTTP basic auth from + it connects to an OpenSearch endpoint that isn't Amazon OpenSearch Serverless + (AOSS) instead: HTTP basic auth from GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no - auth. The AWS session and region are never touched on the local path. + auth. The AWS session and region are never touched on this path. Args: endpoint: The URL of the OpenSearch cluster endpoint. is_local: bool - True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. + True for an OpenSearch endpoint that isn't Amazon OpenSearch Serverless + (AOSS); skips SigV4. **kwargs: Optional parameters for customizing the AsyncOpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. @@ -264,9 +268,10 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, cl index_name: The name of the index to check for existence. dimensions: The number of dimensions for the knn_vector in the index. writeable: Flag indicating whether to create the index if it does not exist. - is_local: True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4. + is_local: True for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS); skips SigV4. client_kwargs: Optional dict of keyword arguments passed through to the - OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP local + OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP endpoint). ``pool_maxsize`` defaults to 1 unless overridden here. Returns: @@ -341,7 +346,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, cl return index_exists -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None, is_local=False): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_local=False, client_kwargs=None): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -354,14 +359,14 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode index_name: Name of the index to be used for storing vectors. dimensions: Dimensions of the vector space used for embeddings. embed_model: Embedding model associated with the vectors. + is_local: True for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS); skips SigV4 and marks the underlying client as non-AOSS + (refresh() is called after bulk ingest). client_kwargs: Optional dict of keyword arguments passed through to both the synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). Defaults that are already set on the clients (``use_ssl``, ``verify_certs``, ``connection_class``, ``timeout``, ``max_retries``, ``retry_on_timeout``) can be overridden here. - is_local: True for a self-managed (non-AWS) OpenSearch endpoint; skips SigV4 - and marks the underlying client as non-AOSS (refresh() is called after - bulk ingest, matching a self-managed cluster's behavior). Returns: OpensearchVectorClient: A configured OpenSearch vector client instance. @@ -508,7 +513,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None, is_local=False): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_local=False, client_kwargs=None): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. @@ -525,11 +530,12 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw the configuration specified in GraphRAGConfig. dimensions (Optional[int]): The dimensions to be used for the embeddings. Defaults to the configuration specified in GraphRAGConfig. + is_local (bool): True for an OpenSearch endpoint that isn't Amazon + OpenSearch Serverless (AOSS); skips AWS SigV4 signing in favor of HTTP + basic auth or no auth. client_kwargs (Optional[Dict[str, Any]]): Optional dict of keyword arguments forwarded to the underlying synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). - is_local (bool): True for a self-managed (non-AWS) OpenSearch endpoint; - skips AWS SigV4 signing in favor of HTTP basic auth or no auth. Returns: OpenSearchIndex: An instance of OpenSearchIndex initialized with the @@ -538,15 +544,15 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw embed_model = embed_model or GraphRAGConfig.embed_model dimensions = dimensions or GraphRAGConfig.embed_dimensions - return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, client_kwargs=client_kwargs, is_local=is_local) + return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_local=is_local, client_kwargs=client_kwargs) endpoint:str index_name:str dimensions:int embed_model:EmbeddingType model_config = ConfigDict(arbitrary_types_allowed=True) - client_kwargs:Optional[Dict[str, Any]]=None is_local:bool=False + client_kwargs:Optional[Dict[str, Any]]=None _client: OpensearchVectorClient = PrivateAttr(default=None) @@ -595,8 +601,8 @@ def client(self) -> OpensearchVectorClient: self.underlying_index_name(), self.dimensions, self.embed_model, - client_kwargs=self.client_kwargs, is_local=self.is_local, + client_kwargs=self.client_kwargs, ) else: self._client = DummyOpensearchVectorClient() From c64e0e093f8fc68abf58e8f0b7bc0fe555f06ab0 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Thu, 16 Jul 2026 15:49:08 -0600 Subject: [PATCH 04/10] refactor(lexical-graph): rename is_local flag to is_serverless for AOSS detection The flag distinguishes Amazon OpenSearch Serverless (AOSS, SigV4 auth) from an OpenSearch endpoint that isn't AOSS (basic auth or no auth), which is what the auth path actually keys off -- not whether the instance is local. An EC2-hosted OpenSearch cluster isn't local but takes the non-serverless path. is_serverless matches AWS terminology and reads at call sites; it defaults to True (AOSS), so aoss:// stays the zero-config default and opensearch:// sets it False. - Invert is_local -> is_serverless across create_os_client/async, index_exists, create_opensearch_vector_client, OpenSearchIndex, and the factory. - Rename the OPENSEARCH_LOCAL constant to OPENSEARCH and _local_http_auth to _basic_http_auth. - Addresses review feedback on #397. --- .../vector/opensearch_vector_index_factory.py | 14 +-- .../vector/opensearch_vector_indexes.py | 86 +++++++++---------- .../storage/vector/test_opensearch_clients.py | 52 +++++------ .../vector/test_vector_index_factories.py | 6 +- 4 files changed, 79 insertions(+), 79 deletions(-) diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index 6ac4186b7..b33c98f21 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -10,7 +10,7 @@ OPENSEARCH_SERVERLESS = 'aoss://' OPENSEARCH_SERVERLESS_DNS = 'aoss.amazonaws.com' -OPENSEARCH_LOCAL = 'opensearch://' +OPENSEARCH = 'opensearch://' class OpenSearchVectorIndexFactory(VectorIndexFactoryMethod): """Factory class for creating OpenSearch vector indexes. @@ -44,23 +44,23 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> names and endpoint configuration, or None if no suitable endpoint is found. """ endpoint = None - is_local = False + is_serverless = True if vector_index_info.startswith(OPENSEARCH_SERVERLESS): endpoint = vector_index_info[len(OPENSEARCH_SERVERLESS):] if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' - elif vector_index_info.startswith(OPENSEARCH_LOCAL): - endpoint = vector_index_info[len(OPENSEARCH_LOCAL):] + elif vector_index_info.startswith(OPENSEARCH): + endpoint = vector_index_info[len(OPENSEARCH):] if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' - is_local = True + is_serverless = False elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info if endpoint: try: from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes import OpenSearchIndex - logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_local: {is_local}]') - return [OpenSearchIndex.for_index(index_name, endpoint, is_local=is_local, **kwargs) for index_name in index_names] + logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_serverless: {is_serverless}]') + return [OpenSearchIndex.for_index(index_name, endpoint, is_serverless=is_serverless, **kwargs) for index_name in index_names] except ImportError as e: raise e diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index adf04d530..036aad014 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -115,7 +115,7 @@ class DummyAuth: DEFAULT_POOL_MAXSIZE = 32 -def _local_http_auth(): +def _basic_http_auth(): """HTTP basic-auth tuple for an OpenSearch endpoint (not Amazon OpenSearch Serverless) from GraphRAGConfig.opensearch_username/opensearch_password, or None if either is unset (an instance with no security plugin configured).""" @@ -127,23 +127,23 @@ def _local_http_auth(): logger.warning('Only one of opensearch_username/opensearch_password is set; connecting without basic auth') return None -def create_os_client(endpoint, is_local=False, **kwargs): +def create_os_client(endpoint, is_serverless=True, **kwargs): """ Creates an OpenSearch client. - By default the client is configured to use AWS Signature Version 4 authentication, - signed with credentials from a pre-configured session and region. When is_local is - True, this instead connects to an OpenSearch endpoint that isn't Amazon OpenSearch - Serverless (AOSS): HTTP basic auth from + By default (is_serverless True) the client targets Amazon OpenSearch Serverless + (AOSS) and authenticates with AWS Signature Version 4, signed with credentials from + a pre-configured session and region. When is_serverless is False, this instead + connects to an OpenSearch endpoint that isn't AOSS: HTTP basic auth from GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise - no auth. The AWS session and region are never touched on this path. + no auth. The AWS session and region are never touched on that path. Args: endpoint: str The OpenSearch endpoint URL to connect to. - is_local: bool - True for an OpenSearch endpoint that isn't Amazon OpenSearch Serverless - (AOSS); skips SigV4. + is_serverless: bool + True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. + False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Any Additional keyword arguments passed to the OpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. @@ -153,8 +153,8 @@ def create_os_client(endpoint, is_local=False, **kwargs): A configured client instance for interacting with OpenSearch. """ - if is_local: - auth = _local_http_auth() + if not is_serverless: + auth = _basic_http_auth() else: session = GraphRAGConfig.session region = GraphRAGConfig.aws_region @@ -179,29 +179,29 @@ def create_os_client(endpoint, is_local=False, **kwargs): **{**defaults, **kwargs}, ) -def create_os_async_client(endpoint, is_local=False, **kwargs): +def create_os_async_client(endpoint, is_serverless=True, **kwargs): """ Creates an asynchronous OpenSearch client. - By default this authenticates with AWS Signature Version 4. When is_local is True, - it connects to an OpenSearch endpoint that isn't Amazon OpenSearch Serverless - (AOSS) instead: HTTP basic auth from + By default (is_serverless True) this targets Amazon OpenSearch Serverless (AOSS) and + authenticates with AWS Signature Version 4. When is_serverless is False, it connects + to an OpenSearch endpoint that isn't AOSS instead: HTTP basic auth from GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no - auth. The AWS session and region are never touched on this path. + auth. The AWS session and region are never touched on that path. Args: endpoint: The URL of the OpenSearch cluster endpoint. - is_local: bool - True for an OpenSearch endpoint that isn't Amazon OpenSearch Serverless - (AOSS); skips SigV4. + is_serverless: bool + True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. + False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Optional parameters for customizing the AsyncOpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. Returns: AsyncOpenSearch: An instantiated asynchronous OpenSearch client. """ - if is_local: - auth = _local_http_auth() + if not is_serverless: + auth = _basic_http_auth() else: session = GraphRAGConfig.session region = GraphRAGConfig.aws_region @@ -254,7 +254,7 @@ def index_is_available(client, index_name): -def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, client_kwargs=None) -> bool: +def index_exists(endpoint, index_name, dimensions, writeable, is_serverless=True, client_kwargs=None) -> bool: """ Checks if an OpenSearch index exists, and optionally creates it if it does not exist. @@ -268,8 +268,8 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, cl index_name: The name of the index to check for existence. dimensions: The number of dimensions for the knn_vector in the index. writeable: Flag indicating whether to create the index if it does not exist. - is_local: True for an OpenSearch endpoint that isn't Amazon OpenSearch - Serverless (AOSS); skips SigV4. + is_serverless: True (default) for Amazon OpenSearch Serverless (AOSS). False for + an OpenSearch endpoint that isn't AOSS; skips SigV4. client_kwargs: Optional dict of keyword arguments passed through to the OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP endpoint). ``pool_maxsize`` defaults to 1 unless overridden here. @@ -278,7 +278,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, cl bool: True if the index exists (or is created successfully), False otherwise. """ client_kwargs = {'pool_maxsize': 1, **(client_kwargs or {})} - client = create_os_client(endpoint, is_local=is_local, **client_kwargs) + client = create_os_client(endpoint, is_serverless=is_serverless, **client_kwargs) embedding_field = 'embedding' @@ -346,7 +346,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_local=False, cl return index_exists -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_local=False, client_kwargs=None): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_serverless=True, client_kwargs=None): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -359,9 +359,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode index_name: Name of the index to be used for storing vectors. dimensions: Dimensions of the vector space used for embeddings. embed_model: Embedding model associated with the vectors. - is_local: True for an OpenSearch endpoint that isn't Amazon OpenSearch - Serverless (AOSS); skips SigV4 and marks the underlying client as non-AOSS - (refresh() is called after bulk ingest). + is_serverless: True (default) for Amazon OpenSearch Serverless (AOSS). False for + an OpenSearch endpoint that isn't AOSS; skips SigV4 and marks the underlying + client as non-AOSS (refresh() is called after bulk ingest). client_kwargs: Optional dict of keyword arguments passed through to both the synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). Defaults that are already set @@ -380,7 +380,7 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode client_kwargs = client_kwargs or {} - logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_local={is_local}]') + logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_serverless={is_serverless}]') client = None retry_count = 0 @@ -392,9 +392,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode dimensions, embedding_field=embedding_field, text_field=text_field, - os_client=create_os_client(endpoint, is_local=is_local, **client_kwargs), - os_async_client=create_os_async_client(endpoint, is_local=is_local, **client_kwargs), - http_auth=DummyAuth(service='local') if is_local else DummyAuth(service='aoss') + os_client=create_os_client(endpoint, is_serverless=is_serverless, **client_kwargs), + os_async_client=create_os_async_client(endpoint, is_serverless=is_serverless, **client_kwargs), + http_auth=DummyAuth(service='aoss') if is_serverless else DummyAuth(service='opensearch') ) start = time.time() @@ -513,7 +513,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_local=False, client_kwargs=None): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_serverless=True, client_kwargs=None): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. @@ -530,9 +530,9 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_local= the configuration specified in GraphRAGConfig. dimensions (Optional[int]): The dimensions to be used for the embeddings. Defaults to the configuration specified in GraphRAGConfig. - is_local (bool): True for an OpenSearch endpoint that isn't Amazon - OpenSearch Serverless (AOSS); skips AWS SigV4 signing in favor of HTTP - basic auth or no auth. + is_serverless (bool): True (default) for Amazon OpenSearch Serverless + (AOSS). False for an OpenSearch endpoint that isn't AOSS; skips AWS SigV4 + signing in favor of HTTP basic auth or no auth. client_kwargs (Optional[Dict[str, Any]]): Optional dict of keyword arguments forwarded to the underlying synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). @@ -544,14 +544,14 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_local= embed_model = embed_model or GraphRAGConfig.embed_model dimensions = dimensions or GraphRAGConfig.embed_dimensions - return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_local=is_local, client_kwargs=client_kwargs) + return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_serverless=is_serverless, client_kwargs=client_kwargs) endpoint:str index_name:str dimensions:int embed_model:EmbeddingType model_config = ConfigDict(arbitrary_types_allowed=True) - is_local:bool=False + is_serverless:bool=True client_kwargs:Optional[Dict[str, Any]]=None _client: OpensearchVectorClient = PrivateAttr(default=None) @@ -595,13 +595,13 @@ def client(self) -> OpensearchVectorClient: self._client = None if not self._client: - if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_local=self.is_local, client_kwargs=self.client_kwargs): + if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_serverless=self.is_serverless, client_kwargs=self.client_kwargs): self._client = create_opensearch_vector_client( self.endpoint, self.underlying_index_name(), self.dimensions, self.embed_model, - is_local=self.is_local, + is_serverless=self.is_serverless, client_kwargs=self.client_kwargs, ) else: @@ -609,7 +609,7 @@ def client(self) -> OpensearchVectorClient: return self._client def index_exists(self): - return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_local=self.is_local, client_kwargs=self.client_kwargs) + return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_serverless=self.is_serverless, client_kwargs=self.client_kwargs) def _clean_id(self, s): """ diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index 510a094ce..2eef99080 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -8,7 +8,7 @@ - create_os_async_client (returns AsyncOpenSearch instance) - create_opensearch_vector_client (success, NotFoundError retry) - OpenSearchIndex.client property (index exists, index does not exist) - - create_os_client/create_os_async_client is_local path (basic auth, no auth, no SigV4) + - create_os_client/create_os_async_client is_serverless=False path (basic auth, no auth, no SigV4) """ import sys @@ -239,7 +239,7 @@ def test_client_creates_real_client_when_index_exists(self): # --------------------------------------------------------------------------- -# create_os_client / create_os_async_client: is_local path +# create_os_client / create_os_async_client: is_serverless=False path # --------------------------------------------------------------------------- class TestCreateOsClientLocal: @@ -251,7 +251,7 @@ def test_local_with_credentials_uses_basic_auth(self): mock_cfg.opensearch_username = "admin" mock_cfg.opensearch_password = "secret" with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: - result = ovi.create_os_client("https://localhost:9200", is_local=True) + result = ovi.create_os_client("https://localhost:9200", is_serverless=False) assert result is mock_os_instance _, kwargs = mock_os.call_args assert kwargs["http_auth"] == ("admin", "secret") @@ -263,7 +263,7 @@ def test_local_without_credentials_uses_no_auth(self): mock_cfg.opensearch_username = None mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: - ovi.create_os_client("https://localhost:9200", is_local=True) + ovi.create_os_client("https://localhost:9200", is_serverless=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None @@ -273,7 +273,7 @@ def test_local_with_only_username_warns_and_uses_no_auth(self, caplog): mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch") as mock_os: with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): - ovi.create_os_client("https://localhost:9200", is_local=True) + ovi.create_os_client("https://localhost:9200", is_serverless=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None assert any("Only one of" in r.message for r in caplog.records) @@ -284,7 +284,7 @@ def test_local_with_only_password_warns_and_uses_no_auth(self, caplog): mock_cfg.opensearch_password = "secret" with patch.object(ovi, "OpenSearch") as mock_os: with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): - ovi.create_os_client("https://localhost:9200", is_local=True) + ovi.create_os_client("https://localhost:9200", is_serverless=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None assert any("Only one of" in r.message for r in caplog.records) @@ -295,11 +295,11 @@ def test_local_never_touches_aws_session_or_sigv4(self): mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch"): with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: - ovi.create_os_client("https://localhost:9200", is_local=True) + ovi.create_os_client("https://localhost:9200", is_serverless=False) mock_sigv4.assert_not_called() mock_cfg.session.assert_not_called() - def test_default_is_local_false_preserves_sigv4(self): + def test_default_is_serverless_true_preserves_sigv4(self): mock_session = MagicMock() mock_session.get_credentials.return_value = MagicMock() @@ -321,7 +321,7 @@ def test_local_with_credentials_uses_basic_auth(self): mock_cfg.opensearch_username = "admin" mock_cfg.opensearch_password = "secret" with patch.object(ovi, "AsyncOpenSearch", return_value=mock_async_instance) as mock_async: - result = ovi.create_os_async_client("https://localhost:9200", is_local=True) + result = ovi.create_os_async_client("https://localhost:9200", is_serverless=False) assert result is mock_async_instance _, kwargs = mock_async.call_args assert kwargs["http_auth"] == ("admin", "secret") @@ -331,7 +331,7 @@ def test_local_without_credentials_uses_no_auth(self): mock_cfg.opensearch_username = None mock_cfg.opensearch_password = None with patch.object(ovi, "AsyncOpenSearch") as mock_async: - ovi.create_os_async_client("https://localhost:9200", is_local=True) + ovi.create_os_async_client("https://localhost:9200", is_serverless=False) _, kwargs = mock_async.call_args assert kwargs["http_auth"] is None @@ -341,25 +341,25 @@ def test_local_never_touches_aws_session_or_sigv4(self): mock_cfg.opensearch_password = None with patch.object(ovi, "AsyncOpenSearch"): with patch.object(ovi, "AWSV4SignerAsyncAuth") as mock_sigv4: - ovi.create_os_async_client("https://localhost:9200", is_local=True) + ovi.create_os_async_client("https://localhost:9200", is_serverless=False) mock_sigv4.assert_not_called() mock_cfg.session.assert_not_called() class TestCreateOpensearchVectorClientLocal: - def test_is_local_threaded_to_both_clients(self): + def test_is_serverless_threaded_to_both_clients(self): with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): with patch.object(ovi, "create_os_client", return_value=MagicMock()) as mock_sync: with patch.object(ovi, "create_os_async_client", return_value=MagicMock()) as mock_async: with patch.object(ovi, "index_is_available", return_value=True): ovi.create_opensearch_vector_client( - "https://localhost:9200", "my-index", 1024, MagicMock(), is_local=True + "https://localhost:9200", "my-index", 1024, MagicMock(), is_serverless=False ) - assert mock_sync.call_args.kwargs.get("is_local") is True - assert mock_async.call_args.kwargs.get("is_local") is True + assert mock_sync.call_args.kwargs.get("is_serverless") is False + assert mock_async.call_args.kwargs.get("is_serverless") is False - def test_is_local_sets_non_aoss_dummy_auth_service(self): + def test_non_serverless_sets_non_aoss_dummy_auth_service(self): captured = {} def fake_vector_client(*args, **kwargs): @@ -371,11 +371,11 @@ def fake_vector_client(*args, **kwargs): with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): with patch.object(ovi, "index_is_available", return_value=True): ovi.create_opensearch_vector_client( - "https://localhost:9200", "my-index", 1024, MagicMock(), is_local=True + "https://localhost:9200", "my-index", 1024, MagicMock(), is_serverless=False ) assert captured["http_auth"].service != "aoss" - def test_default_is_local_false_keeps_aoss_dummy_auth_service(self): + def test_default_is_serverless_true_keeps_aoss_dummy_auth_service(self): captured = {} def fake_vector_client(*args, **kwargs): @@ -403,25 +403,25 @@ def _make_local_index(self): embed_model=MagicMock(), tenant_id=TenantId(), writeable=False, - is_local=True, + is_serverless=False, _client=None, ) - def test_client_property_threads_is_local_to_index_exists(self): + def test_client_property_threads_is_serverless_to_index_exists(self): index = self._make_local_index() with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: index.client - assert mock_index_exists.call_args.kwargs.get("is_local") is True + assert mock_index_exists.call_args.kwargs.get("is_serverless") is False - def test_client_property_threads_is_local_to_vector_client(self): + def test_client_property_threads_is_serverless_to_vector_client(self): index = self._make_local_index() mock_vector_client = MagicMock() with patch.object(ovi, "index_exists", return_value=True): with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client) as mock_create: index.client - assert mock_create.call_args.kwargs.get("is_local") is True + assert mock_create.call_args.kwargs.get("is_serverless") is False def test_client_property_threads_client_kwargs_to_index_exists(self): from graphrag_toolkit.lexical_graph.tenant_id import TenantId @@ -432,7 +432,7 @@ def test_client_property_threads_client_kwargs_to_index_exists(self): embed_model=MagicMock(), tenant_id=TenantId(), writeable=False, - is_local=True, + is_serverless=False, client_kwargs={"use_ssl": False, "verify_certs": False}, _client=None, ) @@ -462,13 +462,13 @@ def test_default_client_kwargs_none_still_sets_pool_maxsize(self): def test_client_kwargs_forwarded_to_create_os_client(self): with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: ovi.index_exists( - "http://localhost:9200", "my-index", 1024, writeable=False, is_local=True, + "http://localhost:9200", "my-index", 1024, writeable=False, is_serverless=False, client_kwargs={"use_ssl": False, "verify_certs": False}, ) _, kwargs = mock_create.call_args assert kwargs["use_ssl"] is False assert kwargs["verify_certs"] is False - assert kwargs["is_local"] is True + assert kwargs["is_serverless"] is False def test_client_kwargs_can_override_default_pool_maxsize(self): with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index 9e3fe27d7..e4542877e 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -87,7 +87,7 @@ def test_try_create_opensearch_prefix_returns_indexes(self): assert result is not None assert len(result) == 2 _, kwargs = mock_cls.for_index.call_args - assert kwargs.get("is_local") is True + assert kwargs.get("is_serverless") is False def test_try_create_opensearch_prefix_defaults_to_https(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory @@ -121,7 +121,7 @@ def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): args, _ = mock_cls.for_index.call_args assert args[1] == "http://localhost:9200" - def test_try_create_aoss_prefix_does_not_set_is_local(self): + def test_try_create_aoss_prefix_sets_is_serverless_true(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory mock_cls = MagicMock() @@ -135,7 +135,7 @@ def test_try_create_aoss_prefix_does_not_set_is_local(self): factory = OpenSearchVectorIndexFactory() factory.try_create(["chunk"], "aoss://https://abc.us-east-1.aoss.amazonaws.com") _, kwargs = mock_cls.for_index.call_args - assert kwargs.get("is_local") is not True + assert kwargs.get("is_serverless") is True # --------------------------------------------------------------------------- From 6ef955ec2572385ae674a80fb68bf6b772a677d3 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Thu, 16 Jul 2026 15:49:08 -0600 Subject: [PATCH 05/10] docs(lexical-graph): document plain-HTTP, self-signed TLS, and non-AWS-auth scope for opensearch:// Note that opensearch:// targets instances without AWS authentication (basic auth or no auth) and that a managed Amazon OpenSearch Service domain using IAM auth is not supported through this prefix yet. Document plain-HTTP endpoints (opensearch://http://...) and disabling certificate verification for self-signed TLS via client_kwargs={'verify_certs': False}. Addresses review feedback on #397. --- .../vector-store-opensearch-serverless.mdx | 13 +++++++++++++ lexical-graph/README.md | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index 90b3f6080..aa6627991 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -149,3 +149,16 @@ with VectorStoreFactory.for_vector_store( ``` Unlike `aoss://` (Amazon OpenSearch Serverless), an `opensearch://` endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. If you omit the scheme from the endpoint (e.g. `opensearch://localhost:9200`), it defaults to `https://`. + +The `opensearch://` prefix targets OpenSearch instances that don't use AWS authentication (basic auth or no auth). A managed Amazon OpenSearch Service domain that authenticates with IAM is not supported through this prefix yet. + +For a local Docker or Finch container over plain HTTP, use `opensearch://http://localhost:9200`. For an endpoint with a self-signed TLS certificate, disable certificate verification by passing `client_kwargs`: + +```python +with VectorStoreFactory.for_vector_store( + 'opensearch://https://localhost:9200', + index_names=['chunk'], + client_kwargs={'verify_certs': False} +) as vector_store: + ... +``` diff --git a/lexical-graph/README.md b/lexical-graph/README.md index f4eb765b0..c18896be2 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -75,7 +75,7 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | | Dummy / no-op | `None` or any unrecognised string — falls back to `DummyGraphStore` / `DummyVectorIndex` | -An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://` (Amazon OpenSearch Serverless), it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. +An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://` (Amazon OpenSearch Serverless), it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. For a self-signed TLS certificate, pass `client_kwargs={'verify_certs': False}` to `VectorStoreFactory.for_vector_store`. The `opensearch://` prefix targets instances without AWS authentication; a managed Amazon OpenSearch Service domain using IAM auth is not supported through this prefix yet. ## Example of use From 9c80c3c4b11d3eeb0df5fa9d03c1c2031e04d857 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Fri, 17 Jul 2026 09:36:26 -0600 Subject: [PATCH 06/10] refactor(lexical-graph): name the AOSS auth flag is_sigv4_auth Follows review on #397: the flag selects AWS SigV4 auth vs basic/no auth, so is_sigv4_auth names the mechanism directly at call sites. Same truth values as before (True for AOSS), no behavior change. --- .../vector/opensearch_vector_index_factory.py | 8 +-- .../vector/opensearch_vector_indexes.py | 52 +++++++++---------- .../storage/vector/test_opensearch_clients.py | 52 +++++++++---------- .../vector/test_vector_index_factories.py | 6 +-- 4 files changed, 59 insertions(+), 59 deletions(-) diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index b33c98f21..538ec2652 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -44,7 +44,7 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> names and endpoint configuration, or None if no suitable endpoint is found. """ endpoint = None - is_serverless = True + is_sigv4_auth = True if vector_index_info.startswith(OPENSEARCH_SERVERLESS): endpoint = vector_index_info[len(OPENSEARCH_SERVERLESS):] if not endpoint.startswith('https://') and not endpoint.startswith('http://'): @@ -53,14 +53,14 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> endpoint = vector_index_info[len(OPENSEARCH):] if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' - is_serverless = False + is_sigv4_auth = False elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info if endpoint: try: from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes import OpenSearchIndex - logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_serverless: {is_serverless}]') - return [OpenSearchIndex.for_index(index_name, endpoint, is_serverless=is_serverless, **kwargs) for index_name in index_names] + logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_sigv4_auth: {is_sigv4_auth}]') + return [OpenSearchIndex.for_index(index_name, endpoint, is_sigv4_auth=is_sigv4_auth, **kwargs) for index_name in index_names] except ImportError as e: raise e diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index 036aad014..2ddedff47 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -127,13 +127,13 @@ def _basic_http_auth(): logger.warning('Only one of opensearch_username/opensearch_password is set; connecting without basic auth') return None -def create_os_client(endpoint, is_serverless=True, **kwargs): +def create_os_client(endpoint, is_sigv4_auth=True, **kwargs): """ Creates an OpenSearch client. - By default (is_serverless True) the client targets Amazon OpenSearch Serverless + By default (is_sigv4_auth True) the client targets Amazon OpenSearch Serverless (AOSS) and authenticates with AWS Signature Version 4, signed with credentials from - a pre-configured session and region. When is_serverless is False, this instead + a pre-configured session and region. When is_sigv4_auth is False, this instead connects to an OpenSearch endpoint that isn't AOSS: HTTP basic auth from GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no auth. The AWS session and region are never touched on that path. @@ -141,7 +141,7 @@ def create_os_client(endpoint, is_serverless=True, **kwargs): Args: endpoint: str The OpenSearch endpoint URL to connect to. - is_serverless: bool + is_sigv4_auth: bool True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Any @@ -153,7 +153,7 @@ def create_os_client(endpoint, is_serverless=True, **kwargs): A configured client instance for interacting with OpenSearch. """ - if not is_serverless: + if not is_sigv4_auth: auth = _basic_http_auth() else: session = GraphRAGConfig.session @@ -179,19 +179,19 @@ def create_os_client(endpoint, is_serverless=True, **kwargs): **{**defaults, **kwargs}, ) -def create_os_async_client(endpoint, is_serverless=True, **kwargs): +def create_os_async_client(endpoint, is_sigv4_auth=True, **kwargs): """ Creates an asynchronous OpenSearch client. - By default (is_serverless True) this targets Amazon OpenSearch Serverless (AOSS) and - authenticates with AWS Signature Version 4. When is_serverless is False, it connects + By default (is_sigv4_auth True) this targets Amazon OpenSearch Serverless (AOSS) and + authenticates with AWS Signature Version 4. When is_sigv4_auth is False, it connects to an OpenSearch endpoint that isn't AOSS instead: HTTP basic auth from GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no auth. The AWS session and region are never touched on that path. Args: endpoint: The URL of the OpenSearch cluster endpoint. - is_serverless: bool + is_sigv4_auth: bool True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Optional parameters for customizing the AsyncOpenSearch client. @@ -200,7 +200,7 @@ def create_os_async_client(endpoint, is_serverless=True, **kwargs): Returns: AsyncOpenSearch: An instantiated asynchronous OpenSearch client. """ - if not is_serverless: + if not is_sigv4_auth: auth = _basic_http_auth() else: session = GraphRAGConfig.session @@ -254,7 +254,7 @@ def index_is_available(client, index_name): -def index_exists(endpoint, index_name, dimensions, writeable, is_serverless=True, client_kwargs=None) -> bool: +def index_exists(endpoint, index_name, dimensions, writeable, is_sigv4_auth=True, client_kwargs=None) -> bool: """ Checks if an OpenSearch index exists, and optionally creates it if it does not exist. @@ -268,7 +268,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_serverless=True index_name: The name of the index to check for existence. dimensions: The number of dimensions for the knn_vector in the index. writeable: Flag indicating whether to create the index if it does not exist. - is_serverless: True (default) for Amazon OpenSearch Serverless (AOSS). False for + is_sigv4_auth: True (default) for Amazon OpenSearch Serverless (AOSS). False for an OpenSearch endpoint that isn't AOSS; skips SigV4. client_kwargs: Optional dict of keyword arguments passed through to the OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP @@ -278,7 +278,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_serverless=True bool: True if the index exists (or is created successfully), False otherwise. """ client_kwargs = {'pool_maxsize': 1, **(client_kwargs or {})} - client = create_os_client(endpoint, is_serverless=is_serverless, **client_kwargs) + client = create_os_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs) embedding_field = 'embedding' @@ -346,7 +346,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_serverless=True return index_exists -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_serverless=True, client_kwargs=None): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_sigv4_auth=True, client_kwargs=None): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -359,7 +359,7 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode index_name: Name of the index to be used for storing vectors. dimensions: Dimensions of the vector space used for embeddings. embed_model: Embedding model associated with the vectors. - is_serverless: True (default) for Amazon OpenSearch Serverless (AOSS). False for + is_sigv4_auth: True (default) for Amazon OpenSearch Serverless (AOSS). False for an OpenSearch endpoint that isn't AOSS; skips SigV4 and marks the underlying client as non-AOSS (refresh() is called after bulk ingest). client_kwargs: Optional dict of keyword arguments passed through to both the @@ -380,7 +380,7 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode client_kwargs = client_kwargs or {} - logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_serverless={is_serverless}]') + logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_sigv4_auth={is_sigv4_auth}]') client = None retry_count = 0 @@ -392,9 +392,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode dimensions, embedding_field=embedding_field, text_field=text_field, - os_client=create_os_client(endpoint, is_serverless=is_serverless, **client_kwargs), - os_async_client=create_os_async_client(endpoint, is_serverless=is_serverless, **client_kwargs), - http_auth=DummyAuth(service='aoss') if is_serverless else DummyAuth(service='opensearch') + os_client=create_os_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs), + os_async_client=create_os_async_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs), + http_auth=DummyAuth(service='aoss') if is_sigv4_auth else DummyAuth(service='opensearch') ) start = time.time() @@ -513,7 +513,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_serverless=True, client_kwargs=None): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_sigv4_auth=True, client_kwargs=None): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. @@ -530,7 +530,7 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_server the configuration specified in GraphRAGConfig. dimensions (Optional[int]): The dimensions to be used for the embeddings. Defaults to the configuration specified in GraphRAGConfig. - is_serverless (bool): True (default) for Amazon OpenSearch Serverless + is_sigv4_auth (bool): True (default) for Amazon OpenSearch Serverless (AOSS). False for an OpenSearch endpoint that isn't AOSS; skips AWS SigV4 signing in favor of HTTP basic auth or no auth. client_kwargs (Optional[Dict[str, Any]]): Optional dict of keyword arguments @@ -544,14 +544,14 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_server embed_model = embed_model or GraphRAGConfig.embed_model dimensions = dimensions or GraphRAGConfig.embed_dimensions - return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_serverless=is_serverless, client_kwargs=client_kwargs) + return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_sigv4_auth=is_sigv4_auth, client_kwargs=client_kwargs) endpoint:str index_name:str dimensions:int embed_model:EmbeddingType model_config = ConfigDict(arbitrary_types_allowed=True) - is_serverless:bool=True + is_sigv4_auth:bool=True client_kwargs:Optional[Dict[str, Any]]=None _client: OpensearchVectorClient = PrivateAttr(default=None) @@ -595,13 +595,13 @@ def client(self) -> OpensearchVectorClient: self._client = None if not self._client: - if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_serverless=self.is_serverless, client_kwargs=self.client_kwargs): + if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs): self._client = create_opensearch_vector_client( self.endpoint, self.underlying_index_name(), self.dimensions, self.embed_model, - is_serverless=self.is_serverless, + is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs, ) else: @@ -609,7 +609,7 @@ def client(self) -> OpensearchVectorClient: return self._client def index_exists(self): - return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_serverless=self.is_serverless, client_kwargs=self.client_kwargs) + return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs) def _clean_id(self, s): """ diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index 2eef99080..ba61d745c 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -8,7 +8,7 @@ - create_os_async_client (returns AsyncOpenSearch instance) - create_opensearch_vector_client (success, NotFoundError retry) - OpenSearchIndex.client property (index exists, index does not exist) - - create_os_client/create_os_async_client is_serverless=False path (basic auth, no auth, no SigV4) + - create_os_client/create_os_async_client is_sigv4_auth=False path (basic auth, no auth, no SigV4) """ import sys @@ -239,7 +239,7 @@ def test_client_creates_real_client_when_index_exists(self): # --------------------------------------------------------------------------- -# create_os_client / create_os_async_client: is_serverless=False path +# create_os_client / create_os_async_client: is_sigv4_auth=False path # --------------------------------------------------------------------------- class TestCreateOsClientLocal: @@ -251,7 +251,7 @@ def test_local_with_credentials_uses_basic_auth(self): mock_cfg.opensearch_username = "admin" mock_cfg.opensearch_password = "secret" with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: - result = ovi.create_os_client("https://localhost:9200", is_serverless=False) + result = ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) assert result is mock_os_instance _, kwargs = mock_os.call_args assert kwargs["http_auth"] == ("admin", "secret") @@ -263,7 +263,7 @@ def test_local_without_credentials_uses_no_auth(self): mock_cfg.opensearch_username = None mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: - ovi.create_os_client("https://localhost:9200", is_serverless=False) + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None @@ -273,7 +273,7 @@ def test_local_with_only_username_warns_and_uses_no_auth(self, caplog): mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch") as mock_os: with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): - ovi.create_os_client("https://localhost:9200", is_serverless=False) + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None assert any("Only one of" in r.message for r in caplog.records) @@ -284,7 +284,7 @@ def test_local_with_only_password_warns_and_uses_no_auth(self, caplog): mock_cfg.opensearch_password = "secret" with patch.object(ovi, "OpenSearch") as mock_os: with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): - ovi.create_os_client("https://localhost:9200", is_serverless=False) + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) _, kwargs = mock_os.call_args assert kwargs["http_auth"] is None assert any("Only one of" in r.message for r in caplog.records) @@ -295,11 +295,11 @@ def test_local_never_touches_aws_session_or_sigv4(self): mock_cfg.opensearch_password = None with patch.object(ovi, "OpenSearch"): with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: - ovi.create_os_client("https://localhost:9200", is_serverless=False) + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) mock_sigv4.assert_not_called() mock_cfg.session.assert_not_called() - def test_default_is_serverless_true_preserves_sigv4(self): + def test_default_is_sigv4_auth_true_preserves_sigv4(self): mock_session = MagicMock() mock_session.get_credentials.return_value = MagicMock() @@ -321,7 +321,7 @@ def test_local_with_credentials_uses_basic_auth(self): mock_cfg.opensearch_username = "admin" mock_cfg.opensearch_password = "secret" with patch.object(ovi, "AsyncOpenSearch", return_value=mock_async_instance) as mock_async: - result = ovi.create_os_async_client("https://localhost:9200", is_serverless=False) + result = ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) assert result is mock_async_instance _, kwargs = mock_async.call_args assert kwargs["http_auth"] == ("admin", "secret") @@ -331,7 +331,7 @@ def test_local_without_credentials_uses_no_auth(self): mock_cfg.opensearch_username = None mock_cfg.opensearch_password = None with patch.object(ovi, "AsyncOpenSearch") as mock_async: - ovi.create_os_async_client("https://localhost:9200", is_serverless=False) + ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) _, kwargs = mock_async.call_args assert kwargs["http_auth"] is None @@ -341,25 +341,25 @@ def test_local_never_touches_aws_session_or_sigv4(self): mock_cfg.opensearch_password = None with patch.object(ovi, "AsyncOpenSearch"): with patch.object(ovi, "AWSV4SignerAsyncAuth") as mock_sigv4: - ovi.create_os_async_client("https://localhost:9200", is_serverless=False) + ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) mock_sigv4.assert_not_called() mock_cfg.session.assert_not_called() class TestCreateOpensearchVectorClientLocal: - def test_is_serverless_threaded_to_both_clients(self): + def test_is_sigv4_auth_threaded_to_both_clients(self): with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): with patch.object(ovi, "create_os_client", return_value=MagicMock()) as mock_sync: with patch.object(ovi, "create_os_async_client", return_value=MagicMock()) as mock_async: with patch.object(ovi, "index_is_available", return_value=True): ovi.create_opensearch_vector_client( - "https://localhost:9200", "my-index", 1024, MagicMock(), is_serverless=False + "https://localhost:9200", "my-index", 1024, MagicMock(), is_sigv4_auth=False ) - assert mock_sync.call_args.kwargs.get("is_serverless") is False - assert mock_async.call_args.kwargs.get("is_serverless") is False + assert mock_sync.call_args.kwargs.get("is_sigv4_auth") is False + assert mock_async.call_args.kwargs.get("is_sigv4_auth") is False - def test_non_serverless_sets_non_aoss_dummy_auth_service(self): + def test_non_sigv4_auth_sets_non_aoss_dummy_auth_service(self): captured = {} def fake_vector_client(*args, **kwargs): @@ -371,11 +371,11 @@ def fake_vector_client(*args, **kwargs): with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): with patch.object(ovi, "index_is_available", return_value=True): ovi.create_opensearch_vector_client( - "https://localhost:9200", "my-index", 1024, MagicMock(), is_serverless=False + "https://localhost:9200", "my-index", 1024, MagicMock(), is_sigv4_auth=False ) assert captured["http_auth"].service != "aoss" - def test_default_is_serverless_true_keeps_aoss_dummy_auth_service(self): + def test_default_is_sigv4_auth_true_keeps_aoss_dummy_auth_service(self): captured = {} def fake_vector_client(*args, **kwargs): @@ -403,25 +403,25 @@ def _make_local_index(self): embed_model=MagicMock(), tenant_id=TenantId(), writeable=False, - is_serverless=False, + is_sigv4_auth=False, _client=None, ) - def test_client_property_threads_is_serverless_to_index_exists(self): + def test_client_property_threads_is_sigv4_auth_to_index_exists(self): index = self._make_local_index() with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: index.client - assert mock_index_exists.call_args.kwargs.get("is_serverless") is False + assert mock_index_exists.call_args.kwargs.get("is_sigv4_auth") is False - def test_client_property_threads_is_serverless_to_vector_client(self): + def test_client_property_threads_is_sigv4_auth_to_vector_client(self): index = self._make_local_index() mock_vector_client = MagicMock() with patch.object(ovi, "index_exists", return_value=True): with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client) as mock_create: index.client - assert mock_create.call_args.kwargs.get("is_serverless") is False + assert mock_create.call_args.kwargs.get("is_sigv4_auth") is False def test_client_property_threads_client_kwargs_to_index_exists(self): from graphrag_toolkit.lexical_graph.tenant_id import TenantId @@ -432,7 +432,7 @@ def test_client_property_threads_client_kwargs_to_index_exists(self): embed_model=MagicMock(), tenant_id=TenantId(), writeable=False, - is_serverless=False, + is_sigv4_auth=False, client_kwargs={"use_ssl": False, "verify_certs": False}, _client=None, ) @@ -462,13 +462,13 @@ def test_default_client_kwargs_none_still_sets_pool_maxsize(self): def test_client_kwargs_forwarded_to_create_os_client(self): with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: ovi.index_exists( - "http://localhost:9200", "my-index", 1024, writeable=False, is_serverless=False, + "http://localhost:9200", "my-index", 1024, writeable=False, is_sigv4_auth=False, client_kwargs={"use_ssl": False, "verify_certs": False}, ) _, kwargs = mock_create.call_args assert kwargs["use_ssl"] is False assert kwargs["verify_certs"] is False - assert kwargs["is_serverless"] is False + assert kwargs["is_sigv4_auth"] is False def test_client_kwargs_can_override_default_pool_maxsize(self): with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index e4542877e..d2aae434f 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -87,7 +87,7 @@ def test_try_create_opensearch_prefix_returns_indexes(self): assert result is not None assert len(result) == 2 _, kwargs = mock_cls.for_index.call_args - assert kwargs.get("is_serverless") is False + assert kwargs.get("is_sigv4_auth") is False def test_try_create_opensearch_prefix_defaults_to_https(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory @@ -121,7 +121,7 @@ def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): args, _ = mock_cls.for_index.call_args assert args[1] == "http://localhost:9200" - def test_try_create_aoss_prefix_sets_is_serverless_true(self): + def test_try_create_aoss_prefix_sets_is_sigv4_auth_true(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory mock_cls = MagicMock() @@ -135,7 +135,7 @@ def test_try_create_aoss_prefix_sets_is_serverless_true(self): factory = OpenSearchVectorIndexFactory() factory.try_create(["chunk"], "aoss://https://abc.us-east-1.aoss.amazonaws.com") _, kwargs = mock_cls.for_index.call_args - assert kwargs.get("is_serverless") is True + assert kwargs.get("is_sigv4_auth") is True # --------------------------------------------------------------------------- From 24a400d0d4678959c0f9f213889b6a27af9b56b8 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Fri, 17 Jul 2026 09:36:26 -0600 Subject: [PATCH 07/10] docs(lexical-graph): refine opensearch:// vector store section per review Rename the section to "Connecting to an OpenSearch vector store", link the opensearch-py client, note OpenSearch's SSL-on default, and mark managed Amazon OpenSearch Service (IAM/SigV4) as not yet supported, tracked in #407. Addresses review on #397. --- .../vector-store-opensearch-serverless.mdx | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index aa6627991..cf766ab5b 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -6,10 +6,10 @@ title: OpenSearch Serverless Vector Store - [Overview](#overview) - [Install dependencies](#install-dependencies) - - [Creating an OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) + - [Creating an Amazon OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) - [Amazon OpenSearch Serverless and custom document IDs](#amazon-opensearch-serverless-and-custom-document-ids) - [Verify and repair an Amazon OpenSearch Serverless vector store](#verify-and-repair-an-amazon-opensearch-serverless-vector-store) - - [OpenSearch](#opensearch) + - [Connecting to an OpenSearch vector store](#connecting-to-an-opensearch-vector-store) ### Overview @@ -132,14 +132,14 @@ Field descriptions: | `num_deleted` | Number of documents deleted from a specific tenant vector index (indicative number only if `dry_run` is `true`) | | `num_unindexed` | Number of nodes that have not been indexed in a specific tenant vector index | -### OpenSearch +### Connecting to an OpenSearch vector store -You can also use OpenSearch directly as a vector store, instead of Amazon OpenSearch Serverless — running locally in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. Supply a connection string that begins `opensearch://`, followed by the endpoint: +You can also use OpenSearch directly as a vector store, instead of Amazon OpenSearch Serverless — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Supply a connection string that begins `opensearch://`, followed by the endpoint: ```python from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory -opensearch_connection_info = 'opensearch://http://localhost:9200' +opensearch_connection_info = 'opensearch://localhost:9200' with VectorStoreFactory.for_vector_store( opensearch_connection_info, @@ -148,9 +148,9 @@ with VectorStoreFactory.for_vector_store( ... ``` -Unlike `aoss://` (Amazon OpenSearch Serverless), an `opensearch://` endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. If you omit the scheme from the endpoint (e.g. `opensearch://localhost:9200`), it defaults to `https://`. +Unlike `aoss://` (Amazon OpenSearch Serverless), an `opensearch://` endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. OpenSearch runs with SSL enabled by default, so if you omit the scheme (e.g. `opensearch://localhost:9200`) the endpoint defaults to `https://`. -The `opensearch://` prefix targets OpenSearch instances that don't use AWS authentication (basic auth or no auth). A managed Amazon OpenSearch Service domain that authenticates with IAM is not supported through this prefix yet. +The `opensearch://` prefix targets OpenSearch instances that don't use AWS authentication (basic auth or no auth). A managed Amazon OpenSearch Service domain that authenticates with IAM is not supported through this prefix yet ([tracked in #407](https://github.com/awslabs/graphrag-toolkit/issues/407)). For a local Docker or Finch container over plain HTTP, use `opensearch://http://localhost:9200`. For an endpoint with a self-signed TLS certificate, disable certificate verification by passing `client_kwargs`: From a2dc39db50767078334e329f1a0496958d6bc4c0 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Fri, 17 Jul 2026 16:56:57 -0600 Subject: [PATCH 08/10] refactor(lexical-graph): address opensearch:// review feedback Pass opensearch:// endpoints through without forcing an https:// scheme. Derive the clients' use_ssl default from the endpoint scheme, since opensearch-py only honors the scheme for https:// and a plain-HTTP endpoint would otherwise still attempt TLS; an explicit use_ssl in client_kwargs still wins. Reject an empty endpoint instead of silently falling through to the dummy vector store. Trim docs that duplicated opensearch-py documentation, link it directly, and add a limitations section referencing #407. Add TODO(#407) markers at the auth branches and extract does_index_exist for readability. --- .../vector-store-opensearch-serverless.mdx | 12 ++++--- lexical-graph/README.md | 2 +- .../vector/opensearch_vector_index_factory.py | 4 +-- .../vector/opensearch_vector_indexes.py | 13 +++++-- .../storage/vector/test_opensearch_clients.py | 36 +++++++++++++++++++ .../vector/test_vector_index_factories.py | 12 +++++-- 6 files changed, 67 insertions(+), 12 deletions(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index cf766ab5b..cbc28ff97 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -134,7 +134,7 @@ Field descriptions: ### Connecting to an OpenSearch vector store -You can also use OpenSearch directly as a vector store, instead of Amazon OpenSearch Serverless — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Supply a connection string that begins `opensearch://`, followed by the endpoint: +You can use [OpenSearch](https://opensearch.org) directly as a vector store — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Supply a connection string that begins with `opensearch://`, followed by the endpoint: ```python from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory @@ -148,11 +148,11 @@ with VectorStoreFactory.for_vector_store( ... ``` -Unlike `aoss://` (Amazon OpenSearch Serverless), an `opensearch://` endpoint never uses AWS SigV4. It connects with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. OpenSearch runs with SSL enabled by default, so if you omit the scheme (e.g. `opensearch://localhost:9200`) the endpoint defaults to `https://`. +The endpoint doesn't need a scheme: `opensearch://localhost:9200` connects over SSL, the toolkit's default (matching OpenSearch, which ships with SSL enabled). Use an explicit `http://` scheme for a plain-HTTP instance: `opensearch://http://localhost:9200`. An endpoint without a port defaults to 9200; for an endpoint on port 443, include the port or an `https://` scheme. -The `opensearch://` prefix targets OpenSearch instances that don't use AWS authentication (basic auth or no auth). A managed Amazon OpenSearch Service domain that authenticates with IAM is not supported through this prefix yet ([tracked in #407](https://github.com/awslabs/graphrag-toolkit/issues/407)). +The client authenticates with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. -For a local Docker or Finch container over plain HTTP, use `opensearch://http://localhost:9200`. For an endpoint with a self-signed TLS certificate, disable certificate verification by passing `client_kwargs`: +Any `client_kwargs` you pass to `VectorStoreFactory.for_vector_store` are forwarded to the opensearch-py client, so all of its [connection options](https://opensearch-project.github.io/opensearch-py/api-ref/clients/opensearch_client.html) — certificate verification, CA bundles, timeouts — are available. For example, for an endpoint with a self-signed TLS certificate: ```python with VectorStoreFactory.for_vector_store( @@ -162,3 +162,7 @@ with VectorStoreFactory.for_vector_store( ) as vector_store: ... ``` + +#### Limitations + +Authentication methods other than basic auth, including SigV4 for IAM-authenticated domains, are not supported yet ([#407](https://github.com/awslabs/graphrag-toolkit/issues/407)). diff --git a/lexical-graph/README.md b/lexical-graph/README.md index c18896be2..95196229f 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -75,7 +75,7 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | | Dummy / no-op | `None` or any unrecognised string — falls back to `DummyGraphStore` / `DummyVectorIndex` | -An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables, or with no auth if neither is set — unlike `aoss://` (Amazon OpenSearch Serverless), it never uses AWS SigV4. `opensearch://localhost:9200` defaults to `https://localhost:9200`; use `opensearch://http://localhost:9200` for a plain-HTTP endpoint. For a self-signed TLS certificate, pass `client_kwargs={'verify_certs': False}` to `VectorStoreFactory.for_vector_store`. The `opensearch://` prefix targets instances without AWS authentication; a managed Amazon OpenSearch Service domain using IAM auth is not supported through this prefix yet. +An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables (no auth if unset), over SSL unless the endpoint has an explicit `http://` scheme. `client_kwargs` passed to `VectorStoreFactory.for_vector_store` are forwarded to the [opensearch-py](https://pypi.org/project/opensearch-py/) client. See [the vector store documentation](https://awslabs.github.io/graphrag-toolkit/lexical-graph/vector-store-opensearch-serverless/) for details and limitations. ## Example of use diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index 538ec2652..ef62fb249 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -51,8 +51,8 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> endpoint = f'https://{endpoint}' elif vector_index_info.startswith(OPENSEARCH): endpoint = vector_index_info[len(OPENSEARCH):] - if not endpoint.startswith('https://') and not endpoint.startswith('http://'): - endpoint = f'https://{endpoint}' + if not endpoint: + raise ValueError(f'Empty endpoint in OpenSearch vector store connection info: {vector_index_info}') is_sigv4_auth = False elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index 2b4057c2b..51b7083b8 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -154,6 +154,7 @@ def create_os_client(endpoint, is_sigv4_auth=True, **kwargs): A configured client instance for interacting with OpenSearch. """ + # TODO(#407): support auth methods beyond SigV4 and basic auth if not is_sigv4_auth: auth = _basic_http_auth() else: @@ -165,7 +166,9 @@ def create_os_client(endpoint, is_sigv4_auth=True, **kwargs): auth = Urllib3AWSV4SignerAuth(credentials, region, service) defaults = { - 'use_ssl': True, + # opensearch-py only honors the endpoint scheme for https://; an http:// + # endpoint would otherwise still connect over TLS via this default. + 'use_ssl': not endpoint.startswith('http://'), 'verify_certs': True, 'connection_class': Urllib3HttpConnection, 'timeout': 300, @@ -201,6 +204,7 @@ def create_os_async_client(endpoint, is_sigv4_auth=True, **kwargs): Returns: AsyncOpenSearch: An instantiated asynchronous OpenSearch client. """ + # TODO(#407): support auth methods beyond SigV4 and basic auth if not is_sigv4_auth: auth = _basic_http_auth() else: @@ -212,7 +216,9 @@ def create_os_async_client(endpoint, is_sigv4_auth=True, **kwargs): auth = AWSV4SignerAsyncAuth(credentials, region, service) defaults = { - 'use_ssl': True, + # opensearch-py only honors the endpoint scheme for https://; an http:// + # endpoint would otherwise still connect over TLS via this default. + 'use_ssl': not endpoint.startswith('http://'), 'verify_certs': True, 'connection_class': AsyncHttpConnection, 'timeout': 300, @@ -699,7 +705,8 @@ def client(self) -> OpensearchVectorClient: self._client = None if not self._client: - if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs): + does_index_exist = index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs) + if does_index_exist: self._client = create_opensearch_vector_client( self.endpoint, self.underlying_index_name(), diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index 431829db0..19e47a018 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -239,6 +239,42 @@ def test_local_with_only_password_warns_and_uses_no_auth(self, caplog): assert kwargs["http_auth"] is None assert any("Only one of" in r.message for r in caplog.records) + def test_schemeless_endpoint_defaults_to_ssl(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + + def test_http_scheme_disables_ssl(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_http_scheme_disables_ssl_async(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_os: + ovi.create_os_async_client("http://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_use_ssl_kwarg_overrides_scheme_default(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False, use_ssl=True) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + def test_local_never_touches_aws_session_or_sigv4(self): with patch.object(ovi, "GraphRAGConfig") as mock_cfg: mock_cfg.opensearch_username = None diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index d2aae434f..ad44df2de 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -89,7 +89,7 @@ def test_try_create_opensearch_prefix_returns_indexes(self): _, kwargs = mock_cls.for_index.call_args assert kwargs.get("is_sigv4_auth") is False - def test_try_create_opensearch_prefix_defaults_to_https(self): + def test_try_create_opensearch_prefix_passes_schemeless_endpoint_through(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory mock_cls = MagicMock() @@ -103,7 +103,7 @@ def test_try_create_opensearch_prefix_defaults_to_https(self): factory = OpenSearchVectorIndexFactory() factory.try_create(["chunk"], "opensearch://localhost:9200") args, _ = mock_cls.for_index.call_args - assert args[1] == "https://localhost:9200" + assert args[1] == "localhost:9200" def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory @@ -121,6 +121,14 @@ def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): args, _ = mock_cls.for_index.call_args assert args[1] == "http://localhost:9200" + def test_try_create_opensearch_prefix_empty_endpoint_raises(self): + import pytest + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + factory = OpenSearchVectorIndexFactory() + with pytest.raises(ValueError, match="Empty endpoint"): + factory.try_create(["chunk"], "opensearch://") + def test_try_create_aoss_prefix_sets_is_sigv4_auth_true(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory From 4ccb438eec97a856cfa1b4f7ae117b2b7c052bfb Mon Sep 17 00:00:00 2001 From: noel-improv Date: Sun, 19 Jul 2026 22:01:01 -0600 Subject: [PATCH 09/10] fix(lexical-graph): harden opensearch:// client construction and TLS handling Address correctness issues in the non-AOSS OpenSearch client path: - Always connect over TLS when signing with SigV4. use_ssl was derived from the endpoint scheme, so an aoss://http://... endpoint would send the signed request (AWS key id and signature) over plaintext. SigV4 now forces TLS regardless of scheme. - Match the http:// scheme case-insensitively. opensearch-py lowercases the scheme, so HTTP://host previously kept use_ssl=True and failed the handshake against a plaintext port. - Warn when credentials would be sent over plain HTTP. - Drop a caller-supplied is_sigv4_auth in the factory before forwarding kwargs, which otherwise collided with the derived keyword and raised TypeError. - Guard the aoss:// branch against an empty endpoint, mirroring the opensearch:// branch, instead of building a client against the bogus host 'https://'. - Make is_sigv4_auth keyword-only in for_index, create_opensearch_vector_client, and index_exists so a positional client_kwargs call keeps forwarding client tuning and the flag can never be passed positionally. - Close the async client when the index is not yet available and the vector client is recreated, so the retry loop no longer leaks AsyncOpenSearch connection pools while the index finishes provisioning. - Document that a default local OpenSearch uses a self-signed certificate, so the quickstart needs verify_certs=False or a trusted CA. --- .../vector-store-opensearch-serverless.mdx | 2 + .../vector/opensearch_vector_index_factory.py | 5 + .../vector/opensearch_vector_indexes.py | 29 +++-- .../storage/vector/test_opensearch_clients.py | 116 ++++++++++++++++++ .../vector/test_vector_index_factories.py | 24 ++++ 5 files changed, 167 insertions(+), 9 deletions(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index cbc28ff97..1e1ce60c6 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -150,6 +150,8 @@ with VectorStoreFactory.for_vector_store( The endpoint doesn't need a scheme: `opensearch://localhost:9200` connects over SSL, the toolkit's default (matching OpenSearch, which ships with SSL enabled). Use an explicit `http://` scheme for a plain-HTTP instance: `opensearch://http://localhost:9200`. An endpoint without a port defaults to 9200; for an endpoint on port 443, include the port or an `https://` scheme. +A default local OpenSearch (Docker or Finch) uses a self-signed certificate, so the example above fails certificate verification until you either trust its CA or disable verification with `client_kwargs={'verify_certs': False}` (see the self-signed example below). Certificate verification stays on by default. + The client authenticates with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. Any `client_kwargs` you pass to `VectorStoreFactory.for_vector_store` are forwarded to the opensearch-py client, so all of its [connection options](https://opensearch-project.github.io/opensearch-py/api-ref/clients/opensearch_client.html) — certificate verification, CA bundles, timeouts — are available. For example, for an endpoint with a self-signed TLS certificate: diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index ef62fb249..72fde98a2 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -47,6 +47,8 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> is_sigv4_auth = True if vector_index_info.startswith(OPENSEARCH_SERVERLESS): endpoint = vector_index_info[len(OPENSEARCH_SERVERLESS):] + if not endpoint: + raise ValueError(f'Empty endpoint in OpenSearch Serverless vector store connection info: {vector_index_info}') if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' elif vector_index_info.startswith(OPENSEARCH): @@ -57,6 +59,9 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info if endpoint: + # is_sigv4_auth is derived from the connection-string scheme above; drop any + # caller-supplied duplicate so it doesn't collide with the keyword below. + kwargs.pop('is_sigv4_auth', None) try: from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes import OpenSearchIndex logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_sigv4_auth: {is_sigv4_auth}]') diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index 51b7083b8..81301f524 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -15,6 +15,7 @@ from llama_index.core.vector_stores.types import VectorStoreQueryResult, VectorStoreQueryMode, MetadataFilters, MetadataFilter from llama_index.core.indices.utils import embed_nodes from llama_index.core.vector_stores.types import MetadataFilters +from llama_index.core.async_utils import asyncio_run from graphrag_toolkit.lexical_graph.metadata import FilterConfig, is_datetime_key, format_datetime from graphrag_toolkit.lexical_graph.versioning import VALID_FROM, VALID_TO, TIMESTAMP_LOWER_BOUND, TIMESTAMP_UPPER_BOUND @@ -165,10 +166,14 @@ def create_os_client(endpoint, is_sigv4_auth=True, **kwargs): auth = Urllib3AWSV4SignerAuth(credentials, region, service) + # SigV4 requests must always go over TLS. On the non-AOSS path honor an explicit + # http:// scheme and default to TLS otherwise. A caller can still override via client_kwargs. + use_ssl = kwargs.get('use_ssl', True if is_sigv4_auth else not endpoint.lower().startswith('http://')) + if auth and not use_ssl: + logger.warning('Connecting with authentication over plain HTTP; credentials will be sent unencrypted') + defaults = { - # opensearch-py only honors the endpoint scheme for https://; an http:// - # endpoint would otherwise still connect over TLS via this default. - 'use_ssl': not endpoint.startswith('http://'), + 'use_ssl': use_ssl, 'verify_certs': True, 'connection_class': Urllib3HttpConnection, 'timeout': 300, @@ -215,10 +220,12 @@ def create_os_async_client(endpoint, is_sigv4_auth=True, **kwargs): auth = AWSV4SignerAsyncAuth(credentials, region, service) + use_ssl = kwargs.get('use_ssl', True if is_sigv4_auth else not endpoint.lower().startswith('http://')) + if auth and not use_ssl: + logger.warning('Connecting with authentication over plain HTTP; credentials will be sent unencrypted') + defaults = { - # opensearch-py only honors the endpoint scheme for https://; an http:// - # endpoint would otherwise still connect over TLS via this default. - 'use_ssl': not endpoint.startswith('http://'), + 'use_ssl': use_ssl, 'verify_certs': True, 'connection_class': AsyncHttpConnection, 'timeout': 300, @@ -399,7 +406,7 @@ def _try_create_index(client, index_name, dimensions, writeable, endpoint, nextg return False -def index_exists(endpoint, index_name, dimensions, writeable, is_sigv4_auth=True, client_kwargs=None) -> bool: +def index_exists(endpoint, index_name, dimensions, writeable, client_kwargs=None, *, is_sigv4_auth=True) -> bool: """ Checks if an OpenSearch index exists, and optionally creates it if it does not exist. @@ -453,7 +460,7 @@ def index_exists(endpoint, index_name, dimensions, writeable, is_sigv4_auth=True client.close() -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, is_sigv4_auth=True, client_kwargs=None): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None, *, is_sigv4_auth=True): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -520,7 +527,11 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode logger.debug(f'Index online [index_name: {index_name}, endpoint: {endpoint}, elapsed_seconds: {elapsed}]') else: logger.debug(f'Index not yet available, recreating client [index_name: {index_name}, endpoint: {endpoint}, elapsed_seconds: {elapsed}]') + # Close both clients before recreating; the async client holds its own + # connection pool, so dropping it without closing would leak on every + # retry while the index finishes provisioning. client._os_client.close() + asyncio_run(client._os_async_client.close()) client = None except NotFoundError as err: @@ -620,7 +631,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, is_sigv4_auth=True, client_kwargs=None): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None, *, is_sigv4_auth=True): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index 19e47a018..51bfeb5e4 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -140,6 +140,49 @@ def side_effect(*args, **kwargs): assert result is mock_vector_client assert call_count["n"] == 2 + def test_closes_async_client_when_not_available_then_recreates(self): + # Index unavailable on the first attempt, available on the second. The first + # attempt's clients must both be closed before recreating, so the AsyncOpenSearch + # connection pool is not leaked while the index finishes provisioning. + first_client = MagicMock() + second_client = MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=[first_client, second_client]): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", side_effect=[False, True]): + # (start, elapsed-init, elapsed-in-loop) per attempt; the 60 forces + # the first attempt's inner wait to exit as not-available. + with patch.object(ovi.time, "time", side_effect=[0, 0, 60, 0, 0, 0]): + with patch.object(ovi.time, "sleep"): + with patch.object(ovi, "asyncio_run") as mock_async_run: + result = ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock() + ) + assert result is second_client + # first attempt's sync client closed directly, async client + # closed via asyncio_run(...close()) — no leak + assert first_client._os_client.close.called + assert mock_async_run.called + + def test_client_kwargs_positional_binds_to_client_kwargs_not_sigv4(self): + captured = {} + + def fake_create_os_client(endpoint, is_sigv4_auth=True, **kwargs): + captured.update(kwargs) + captured["is_sigv4_auth"] = is_sigv4_auth + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): + with patch.object(ovi, "create_os_client", side_effect=fake_create_os_client): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock(), {"timeout": 99} + ) + assert captured.get("timeout") == 99 + assert captured.get("is_sigv4_auth") is True + # --------------------------------------------------------------------------- # OpenSearchIndex.client property @@ -257,6 +300,49 @@ def test_http_scheme_disables_ssl(self): _, kwargs = mock_os.call_args assert kwargs["use_ssl"] is False + def test_uppercase_http_scheme_disables_ssl(self): + # opensearch-py lowercases the scheme, so an uppercase http:// must also + # be detected as plain HTTP rather than defaulting to TLS. + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("HTTP://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_sigv4_auth_forces_ssl_even_for_http_endpoint(self): + # A SigV4-signed request must never go over plaintext, regardless of an + # explicit http:// scheme on the endpoint. + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "OpenSearch") as mock_os: + with patch.object(ovi, "Urllib3AWSV4SignerAuth"): + ovi.create_os_client("http://my-collection.aoss.amazonaws.com", is_sigv4_auth=True) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + + def test_basic_auth_over_plain_http_warns(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False) + assert any("unencrypted" in r.message for r in caplog.records) + + def test_basic_auth_over_https_does_not_warn(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + assert not any("unencrypted" in r.message for r in caplog.records) + def test_http_scheme_disables_ssl_async(self): with patch.object(ovi, "GraphRAGConfig") as mock_cfg: mock_cfg.opensearch_username = None @@ -331,6 +417,36 @@ def test_local_never_touches_aws_session_or_sigv4(self): mock_sigv4.assert_not_called() mock_cfg.session.assert_not_called() + def test_uppercase_http_scheme_disables_ssl_async(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + ovi.create_os_async_client("HTTP://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_async.call_args + assert kwargs["use_ssl"] is False + + def test_sigv4_auth_forces_ssl_even_for_http_endpoint_async(self): + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + with patch.object(ovi, "AWSV4SignerAsyncAuth"): + ovi.create_os_async_client("http://my-collection.aoss.amazonaws.com", is_sigv4_auth=True) + _, kwargs = mock_async.call_args + assert kwargs["use_ssl"] is True + + def test_basic_auth_over_plain_http_warns_async(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "AsyncOpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_async_client("http://localhost:9200", is_sigv4_auth=False) + assert any("unencrypted" in r.message for r in caplog.records) + class TestCreateOpensearchVectorClientLocal: diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index ad44df2de..bfbab9bd0 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -145,6 +145,30 @@ def test_try_create_aoss_prefix_sets_is_sigv4_auth_true(self): _, kwargs = mock_cls.for_index.call_args assert kwargs.get("is_sigv4_auth") is True + def test_try_create_aoss_prefix_empty_endpoint_raises(self): + import pytest + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + factory = OpenSearchVectorIndexFactory() + with pytest.raises(ValueError, match="Empty endpoint"): + factory.try_create(["chunk"], "aoss://") + + def test_try_create_caller_supplied_is_sigv4_auth_does_not_collide(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "opensearch://localhost:9200", is_sigv4_auth=True) + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_sigv4_auth") is False + # --------------------------------------------------------------------------- # PGVectorIndexFactory.try_create From 2027d6ed3ab87fe17dd4aa35e5ba332379b88433 Mon Sep 17 00:00:00 2001 From: noel-improv Date: Mon, 20 Jul 2026 10:26:28 -0600 Subject: [PATCH 10/10] refactor(lexical-graph): route OpenSearch by http(s):// scheme, drop opensearch:// Recognize a self-managed OpenSearch endpoint by its http:// or https:// scheme instead of requiring an opensearch:// prefix. aoss:// (and bare AOSS domains) still route to SigV4; anything else is treated as a non-AOSS cluster. The factory stays gated on the http(s):// scheme rather than claiming every non-aoss:// string, because it runs first in the vector-store dispatch and a blind catch-all would swallow postgres://, neptune-graph://, and s3vectors://. A stale opensearch:// string now falls through to a clear 'Unrecognized vector store info' error rather than mis-routing. Docs and README updated to the scheme-based form; tests cover the http(s):// endpoints, guard that a bare AOSS domain keeps SigV4, and that opensearch:// and schemeless endpoints no longer match. --- .../vector-store-opensearch-serverless.mdx | 8 +-- lexical-graph/README.md | 2 +- .../vector/opensearch_vector_index_factory.py | 11 ++-- .../vector/test_vector_index_factories.py | 51 +++++++++---------- 4 files changed, 35 insertions(+), 37 deletions(-) diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index 1e1ce60c6..c863997de 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -134,12 +134,12 @@ Field descriptions: ### Connecting to an OpenSearch vector store -You can use [OpenSearch](https://opensearch.org) directly as a vector store — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Supply a connection string that begins with `opensearch://`, followed by the endpoint: +You can use [OpenSearch](https://opensearch.org) directly as a vector store — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Pass the endpoint as an `http://` or `https://` URL: ```python from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory -opensearch_connection_info = 'opensearch://localhost:9200' +opensearch_connection_info = 'https://localhost:9200' with VectorStoreFactory.for_vector_store( opensearch_connection_info, @@ -148,7 +148,7 @@ with VectorStoreFactory.for_vector_store( ... ``` -The endpoint doesn't need a scheme: `opensearch://localhost:9200` connects over SSL, the toolkit's default (matching OpenSearch, which ships with SSL enabled). Use an explicit `http://` scheme for a plain-HTTP instance: `opensearch://http://localhost:9200`. An endpoint without a port defaults to 9200; for an endpoint on port 443, include the port or an `https://` scheme. +Any `https://`/`http://` endpoint that isn't an Amazon OpenSearch Serverless domain is treated as a self-managed OpenSearch cluster (use the `aoss://` prefix for Serverless). Use `https://` to connect over SSL, the toolkit's default (matching OpenSearch, which ships with SSL enabled), or `http://` for a plain-HTTP instance. An endpoint without a port defaults to 9200. A default local OpenSearch (Docker or Finch) uses a self-signed certificate, so the example above fails certificate verification until you either trust its CA or disable verification with `client_kwargs={'verify_certs': False}` (see the self-signed example below). Certificate verification stays on by default. @@ -158,7 +158,7 @@ Any `client_kwargs` you pass to `VectorStoreFactory.for_vector_store` are forwar ```python with VectorStoreFactory.for_vector_store( - 'opensearch://https://localhost:9200', + 'https://localhost:9200', index_names=['chunk'], client_kwargs={'verify_certs': False} ) as vector_store: diff --git a/lexical-graph/README.md b/lexical-graph/README.md index 95196229f..813c5ad75 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -69,7 +69,7 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | Neptune Database (graph) | `neptune-db://` or any hostname ending `.neptune.amazonaws.com` | | Neo4j (graph) | `bolt://`, `bolt+ssc://`, `bolt+s://`, `neo4j://`, `neo4j+ssc://`, or `neo4j+s://` URLs | | Amazon OpenSearch Serverless / AOSS (vector) | `aoss://` | -| OpenSearch (vector) | `opensearch://` | +| OpenSearch (vector) | `https://` or `http://` | | Neptune Analytics (vector) | `neptune-graph://` | | pgvector (vector) | constructed via `PGVectorIndexFactory` | | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index 72fde98a2..62254c323 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -10,7 +10,6 @@ OPENSEARCH_SERVERLESS = 'aoss://' OPENSEARCH_SERVERLESS_DNS = 'aoss.amazonaws.com' -OPENSEARCH = 'opensearch://' class OpenSearchVectorIndexFactory(VectorIndexFactoryMethod): """Factory class for creating OpenSearch vector indexes. @@ -51,13 +50,13 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> raise ValueError(f'Empty endpoint in OpenSearch Serverless vector store connection info: {vector_index_info}') if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' - elif vector_index_info.startswith(OPENSEARCH): - endpoint = vector_index_info[len(OPENSEARCH):] - if not endpoint: - raise ValueError(f'Empty endpoint in OpenSearch vector store connection info: {vector_index_info}') - is_sigv4_auth = False elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info + elif vector_index_info.startswith('https://') or vector_index_info.startswith('http://'): + # Any other http(s) endpoint is a self-managed (non-AOSS) OpenSearch cluster, + # so no prefix is required; aoss:// and bare AOSS domains are handled above. + endpoint = vector_index_info + is_sigv4_auth = False if endpoint: # is_sigv4_auth is derived from the connection-string scheme above; drop any # caller-supplied duplicate so it doesn't collide with the keyword below. diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index bfbab9bd0..f9495a144 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -4,7 +4,7 @@ """Tests for vector index factory try_create methods. Covers: - - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, opensearch://, no match) + - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, http(s):// endpoint, no match) - PGVectorIndexFactory.try_create (postgres://, postgresql://, no match) - S3VectorIndexFactory.try_create (s3vectors://, no match) """ @@ -66,8 +66,12 @@ def test_try_create_https_aoss_dns_returns_indexes(self): "https://abc.us-east-1.aoss.amazonaws.com" ) assert result is not None + # a bare AOSS domain must stay on SigV4 — this branch is matched ahead of + # the generic https:// branch, so the ordering is load-bearing + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_sigv4_auth") is True - def test_try_create_opensearch_prefix_returns_indexes(self): + def test_try_create_https_endpoint_returns_indexes(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory mock_index = MagicMock() @@ -82,14 +86,16 @@ def test_try_create_opensearch_prefix_returns_indexes(self): factory = OpenSearchVectorIndexFactory() result = factory.try_create( ["chunk", "statement"], - "opensearch://localhost:9200" + "https://localhost:9200" ) assert result is not None assert len(result) == 2 - _, kwargs = mock_cls.for_index.call_args + args, kwargs = mock_cls.for_index.call_args + # endpoint passed through verbatim, no prefix stripping, non-AOSS auth + assert args[1] == "https://localhost:9200" assert kwargs.get("is_sigv4_auth") is False - def test_try_create_opensearch_prefix_passes_schemeless_endpoint_through(self): + def test_try_create_http_endpoint_passes_through_and_sets_non_sigv4(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory mock_cls = MagicMock() @@ -101,33 +107,26 @@ def test_try_create_opensearch_prefix_passes_schemeless_endpoint_through(self): ) }): factory = OpenSearchVectorIndexFactory() - factory.try_create(["chunk"], "opensearch://localhost:9200") - args, _ = mock_cls.for_index.call_args - assert args[1] == "localhost:9200" + factory.try_create(["chunk"], "http://localhost:9200") + args, kwargs = mock_cls.for_index.call_args + assert args[1] == "http://localhost:9200" + assert kwargs.get("is_sigv4_auth") is False - def test_try_create_opensearch_prefix_preserves_explicit_scheme(self): + def test_try_create_schemeless_endpoint_returns_none(self): + # Without the opensearch:// prefix, a schemeless endpoint is no longer claimed + # by this factory; the caller must supply an http:// or https:// scheme. from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory - mock_cls = MagicMock() - mock_cls.for_index.return_value = MagicMock() - - with patch.dict("sys.modules", { - "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( - OpenSearchIndex=mock_cls - ) - }): - factory = OpenSearchVectorIndexFactory() - factory.try_create(["chunk"], "opensearch://http://localhost:9200") - args, _ = mock_cls.for_index.call_args - assert args[1] == "http://localhost:9200" + factory = OpenSearchVectorIndexFactory() + assert factory.try_create(["chunk"], "localhost:9200") is None - def test_try_create_opensearch_prefix_empty_endpoint_raises(self): - import pytest + def test_try_create_opensearch_prefix_no_longer_matched(self): + # opensearch:// was removed; such a string should fall through (return None) + # so the dispatch reports it as unrecognized rather than mis-routing. from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory factory = OpenSearchVectorIndexFactory() - with pytest.raises(ValueError, match="Empty endpoint"): - factory.try_create(["chunk"], "opensearch://") + assert factory.try_create(["chunk"], "opensearch://localhost:9200") is None def test_try_create_aoss_prefix_sets_is_sigv4_auth_true(self): from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory @@ -165,7 +164,7 @@ def test_try_create_caller_supplied_is_sigv4_auth_does_not_collide(self): ) }): factory = OpenSearchVectorIndexFactory() - factory.try_create(["chunk"], "opensearch://localhost:9200", is_sigv4_auth=True) + factory.try_create(["chunk"], "https://localhost:9200", is_sigv4_auth=True) _, kwargs = mock_cls.for_index.call_args assert kwargs.get("is_sigv4_auth") is False