From 7bd97f8d2a3429f266f3ac7e7a4721dbeaa5b36b Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 23 Sep 2026 12:21:14 +0200 Subject: [PATCH 1/3] feat(python): add Oracle vector store connector --- .../workflows/python-integration-tests.yml | 3 +- .github/workflows/python-merge-tests.yml | 4 +- python/PACKAGE_STATUS.md | 1 + python/packages/oracle/LICENSE | 21 + python/packages/oracle/README.md | 147 +++ .../oracle/agent_framework_oracle/__init__.py | 16 + .../agent_framework_oracle/_vector_store.py | 931 ++++++++++++++++++ .../oracle/agent_framework_oracle/py.typed | 0 python/packages/oracle/pyproject.toml | 63 ++ .../packages/oracle/samples/oracle_vectors.py | 67 ++ .../packages/oracle/tests/oracle/conftest.py | 112 +++ .../oracle/tests/oracle/test_collection.py | 449 +++++++++ .../oracle/tests/oracle/test_filters.py | 130 +++ .../oracle/tests/oracle/test_integration.py | 104 ++ .../oracle/tests/oracle/test_settings.py | 183 ++++ python/pyproject.toml | 1 + python/uv.lock | 48 + 17 files changed, 2278 insertions(+), 2 deletions(-) create mode 100644 python/packages/oracle/LICENSE create mode 100644 python/packages/oracle/README.md create mode 100644 python/packages/oracle/agent_framework_oracle/__init__.py create mode 100644 python/packages/oracle/agent_framework_oracle/_vector_store.py create mode 100644 python/packages/oracle/agent_framework_oracle/py.typed create mode 100644 python/packages/oracle/pyproject.toml create mode 100644 python/packages/oracle/samples/oracle_vectors.py create mode 100644 python/packages/oracle/tests/oracle/conftest.py create mode 100644 python/packages/oracle/tests/oracle/test_collection.py create mode 100644 python/packages/oracle/tests/oracle/test_filters.py create mode 100644 python/packages/oracle/tests/oracle/test_integration.py create mode 100644 python/packages/oracle/tests/oracle/test_settings.py diff --git a/.github/workflows/python-integration-tests.yml b/.github/workflows/python-integration-tests.yml index d85bba7c9f9..4733488d5f8 100644 --- a/.github/workflows/python-integration-tests.yml +++ b/.github/workflows/python-integration-tests.yml @@ -327,13 +327,14 @@ jobs: fallback_url: ${{ env.LOCAL_MCP_URL }} - name: Prefer local MCP URL when available run: echo "LOCAL_MCP_URL=${{ steps.local-mcp.outputs.effective_url }}" >> "$GITHUB_ENV" - - name: Test with pytest (Anthropic, Hyperlight, Mistral, Ollama, MCP integration) + - name: Test with pytest (Anthropic, Hyperlight, Mistral, Ollama, Oracle, MCP integration) run: > uv run pytest --import-mode=importlib packages/anthropic/tests packages/hyperlight/tests packages/mistral/tests packages/ollama/tests + packages/oracle/tests packages/core/tests/core/test_mcp.py packages/hosting-mcp/tests -m integration diff --git a/.github/workflows/python-merge-tests.yml b/.github/workflows/python-merge-tests.yml index 2d75ad02844..3ed9b4039ee 100644 --- a/.github/workflows/python-merge-tests.yml +++ b/.github/workflows/python-merge-tests.yml @@ -73,6 +73,7 @@ jobs: - 'python/packages/hyperlight/**' - 'python/packages/mistral/**' - 'python/packages/ollama/**' + - 'python/packages/oracle/**' - 'python/packages/core/agent_framework/_mcp.py' - 'python/packages/core/tests/core/test_mcp.py' - 'python/packages/hosting-mcp/**' @@ -460,13 +461,14 @@ jobs: fallback_url: ${{ env.LOCAL_MCP_URL }} - name: Prefer local MCP URL when available run: echo "LOCAL_MCP_URL=${{ steps.local-mcp.outputs.effective_url }}" >> "$GITHUB_ENV" - - name: Test with pytest (Anthropic, Hyperlight, Mistral, Ollama, MCP integration) + - name: Test with pytest (Anthropic, Hyperlight, Mistral, Ollama, Oracle, MCP integration) run: > uv run pytest --import-mode=importlib packages/anthropic/tests packages/hyperlight/tests packages/mistral/tests packages/ollama/tests + packages/oracle/tests packages/core/tests/core/test_mcp.py packages/hosting-mcp/tests -m integration diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index 51ff0f2f7a6..d125ff73fae 100644 --- a/python/PACKAGE_STATUS.md +++ b/python/PACKAGE_STATUS.md @@ -48,6 +48,7 @@ Status is grouped into these buckets: | `agent-framework-monty` | `python/packages/monty` | `beta` | | `agent-framework-ollama` | `python/packages/ollama` | `beta` | | `agent-framework-openai` | `python/packages/openai` | `released` | +| `agent-framework-oracle` | `python/packages/oracle` | `alpha` | | `agent-framework-orchestrations` | `python/packages/orchestrations` | `released` | | `agent-framework-postgres` | `python/packages/postgres` | `alpha` | | `agent-framework-purview` | `python/packages/purview` | `beta` | diff --git a/python/packages/oracle/LICENSE b/python/packages/oracle/LICENSE new file mode 100644 index 00000000000..9e841e7a26e --- /dev/null +++ b/python/packages/oracle/LICENSE @@ -0,0 +1,21 @@ + MIT License + + Copyright (c) Microsoft Corporation. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE diff --git a/python/packages/oracle/README.md b/python/packages/oracle/README.md new file mode 100644 index 00000000000..a93b340a2fb --- /dev/null +++ b/python/packages/oracle/README.md @@ -0,0 +1,147 @@ +# Agent Framework Oracle vector store + +An alpha connector for storing and searching native `VECTOR` columns in Oracle +Database 23ai or newer. `OracleCollection` implements async batch CRUD and +vector search; `OracleStore` shares a client across collections; `OracleSettings` +resolves connection configuration through Agent Framework. + +## Installation and setup + +```bash +pip install agent-framework-oracle --pre +``` + +Requires Python 3.10+, Oracle Database 23ai or newer with `COMPATIBLE` set to +**23.4.0 or higher**, `python-oracledb` 2.2.x or 3.x, and permission to +create/drop tables in the connected user's schema. The connector uses the +driver's async Thin-mode API; it does not initialize the Thick client or +provision a database. +It does not create or alter schemas, vector indexes, or existing tables. + +Set `ORACLE_DSN` (for example `localhost:1521/FREEPDB1`), `ORACLE_USER`, and +`ORACLE_PASSWORD`, or pass `dsn`, `user`, and `password` to either constructor. +Credentials are resolved in order: **explicit argument > selected `.env` file +> process environment**. A `.env` file is read only when `env_file_path` is +specified; `env_file_encoding` overrides UTF-8. `password` accepts `str` or +Agent Framework `SecretString` and is unwrapped only when opening a connection. +Missing or empty credentials are errors. + +Alternatively, pass a configured `oracledb.AsyncConnection` or +`oracledb.AsyncConnectionPool` as `client` for advanced authentication. A +supplied client cannot be combined with connection settings and remains +caller-owned. A connector-created pool opens on first use and closes on +`close()` or async context exit. A collection obtained from a store borrows its +client; closing that collection does not close the store. The store must remain +open while its collections are used. + +Writes on pooled connections (both connector-owned and caller-supplied pools) +are committed on success and rolled back on failure. Writes to a supplied +`AsyncConnection` are left in the caller's transaction; **the caller must +commit or roll back**. Oracle DDL commits independently of the surrounding +transaction. Batch upserts on a borrowed connection may leave partial pending +work on failure, so roll it back before reuse. + +`ensure_collection_exists()` leaves an existing table unchanged and does not +validate its schema; use a matching definition or create a new table. It never +migrates data or changes existing indexes. + +## Usage + +```python +import asyncio +from dataclasses import dataclass +from typing import Annotated + +from agent_framework import Filter, VectorStoreField, vectorstoremodel +from agent_framework_oracle import OracleStore + + +@vectorstoremodel(collection_name="articles") +@dataclass +class Article: + id: Annotated[str, VectorStoreField("key")] + text: Annotated[str, VectorStoreField("data")] + embedding: Annotated[list[float] | None, VectorStoreField("vector", dimensions=3)] = None + + +async def main() -> None: + async with OracleStore() as store: + collection = store.get_collection(Article) + await collection.ensure_collection_exists() + await collection.upsert([Article("one", "Oracle vector search", [1, 0, 0])], generate_vectors=False) + results = await collection.search( + vector=[1, 0, 0], filter=Filter("text", "contains_text", "Oracle"), top=3 + ) + async for result in results: + print(result["record"].text, result["score"]) + + +asyncio.run(main()) +``` + +Pass `generate_vectors=False` for precomputed vectors, or configure an +`embedding_generator` for local generation. `get()` excludes vector columns by +default; pass `include_vectors=True` to restore them. `get(keys)` preserves +requested key order and duplicates, omitting missing keys; filtered retrieval +supports `top`, `skip`, and `order_by`. Both `get()` and `search()` execute +portable filters and paging in Oracle. The +[runnable sample](https://github.com/microsoft/agent-framework/blob/main/python/packages/oracle/samples/oracle_vectors.py) +creates and deletes a unique test table. + +## Capabilities and limits + +- A collection is one table in the **current user's schema**. Keys are + application-provided strings (up to 512 UTF-8 bytes), signed 64-bit integers, + or UUIDs; automatically generated keys are not supported. +- Multiple nullable native vector columns are supported. Vector fields may + declare `float`/`float32` (Oracle `FLOAT32`), `float64` (`FLOAT64`), or `int8` + (`INT8`); omitted element types use `FLOAT32`. Dimensions must be 1–65535. + Supplied vectors must be dense finite numeric sequences with the declared + dimensions. Binary and sparse vectors are not supported. +- Scalar data fields support `str` (up to 4000 UTF-8 bytes), `int` (signed + 64-bit), `float` (`BINARY_DOUBLE`), and `bool` (stored as `NUMBER(1,0)`). + Oracle converts empty strings to `NULL`, so this connector rejects empty + string **values** rather than changing their meaning. JSON fields, nested + paths, provider annotations, data/full-text/vector indexes, and schema + migration are not supported. +- Search uses Oracle `VECTOR_DISTANCE` with an explicit metric. The default + is `COSINE` **distance**; `cosine_similarity`, Euclidean distance, dot + product, negative dot product, squared Euclidean distance, and Manhattan + distance are also available. Scores are native + metric values, **not probabilities**. For distances a `score_threshold` is + a maximum; for similarities/dot product it is a minimum. The cutoff is + applied in SQL **before** paging. No ANN indexes are created or managed; + Oracle may use an existing index if the table already has one. There is no + exact/approximate toggle, server-side embedding generation, or + keyword-hybrid search. Index-selected approximate searches may return fewer + than the requested number of results. +- Portable filters support scalar equality (including `NULL` and boolean + distinctions), `in`/`not_in`, numeric ranges, `is_null`, `is_not_null`, + `exists`, literal text `contains_text`/`starts_with`/`ends_with`, and + AND/OR/NOT. Collection membership and nested filters are rejected. + Ordered filters on integer fields reject non-integral float operands. + String comparison follows the configured Oracle collation. Input values + are bound; identifiers are checked and quoted as individual names. + +## Opt-in database tests + +The unit tests require no Oracle server. To run the live integration test, +provide **all three** `ORACLE_TEST_DSN`, `ORACLE_TEST_USER`, and +`ORACLE_TEST_PASSWORD` for an explicitly designated disposable Oracle 23ai+ +schema where the user can create and drop tables: + +```bash +cd python +uv run --package agent-framework-oracle pytest packages/oracle/tests \ + -m integration +``` + +Without those variables, the integration test skips; it never falls back to +ordinary application credentials. The fixture creates a unique table and +deletes only that table. + +## Documentation + +- [Oracle Database AI Vector Search requirements](https://docs.oracle.com/en/database/oracle/oracle-database/26/vecse/overview-ai-vector-search.html) +- [python-oracledb 2.2 vector types](https://python-oracledb.readthedocs.io/en/v2.2.0/user_guide/vector_data_type.html) +- [Microsoft Agent Framework](https://learn.microsoft.com/agent-framework/) diff --git a/python/packages/oracle/agent_framework_oracle/__init__.py b/python/packages/oracle/agent_framework_oracle/__init__.py new file mode 100644 index 00000000000..d3b1af683c4 --- /dev/null +++ b/python/packages/oracle/agent_framework_oracle/__init__.py @@ -0,0 +1,16 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Oracle Database 23ai native-vector collections and stores.""" + +from __future__ import annotations + +import importlib.metadata + +from ._vector_store import OracleCollection, OracleSettings, OracleStore + +try: + __version__ = importlib.metadata.version(__name__) +except importlib.metadata.PackageNotFoundError: + __version__ = "0.0.0" + +__all__ = ["OracleCollection", "OracleSettings", "OracleStore", "__version__"] diff --git a/python/packages/oracle/agent_framework_oracle/_vector_store.py b/python/packages/oracle/agent_framework_oracle/_vector_store.py new file mode 100644 index 00000000000..2f73f3199e6 --- /dev/null +++ b/python/packages/oracle/agent_framework_oracle/_vector_store.py @@ -0,0 +1,931 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Oracle Database 23ai vector collections and stores.""" + +from __future__ import annotations + +import asyncio +import math +from array import array +from collections.abc import AsyncGenerator, Collection, Mapping, Sequence +from contextlib import asynccontextmanager +from decimal import Decimal +from typing import Any, ClassVar, Generic, Protocol, TypeAlias, cast +from uuid import UUID + +import oracledb +from agent_framework import ( + BaseVectorCollection, + BaseVectorSearch, + BaseVectorStore, + FilterGroup, + SearchResults, + SecretString, + VectorStoreCollectionDefinition, + VectorStoreField, + load_settings, +) +from agent_framework._vector_filters import FilterExpression +from agent_framework._vectors import EmbeddingClient, SearchType, Vector +from agent_framework.exceptions import IntegrationException, IntegrationInvalidResponseException +from typing_extensions import Self, TypedDict, TypeVar + +KeyT = TypeVar("KeyT", default=Any) +ModelT = TypeVar("ModelT", default=Any) +OracleClient: TypeAlias = oracledb.AsyncConnection | oracledb.AsyncConnectionPool + + +class _OracleCursor(Protocol): + """Typed boundary for python-oracledb's partially annotated async cursor.""" + + async def __aenter__(self) -> Self: ... + async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: ... + async def execute(self, statement: str, parameters: Mapping[str, Any] | None = None) -> None: ... + async def executemany(self, statement: str, parameters: Sequence[Mapping[str, Any]]) -> None: ... + async def fetchone(self) -> Sequence[Any] | None: ... + async def fetchall(self) -> list[Sequence[Any]]: ... + def setinputsizes(self, **kwargs: Any) -> Any: ... + + +_MAX_IDENTIFIER_BYTES = 128 +_MAX_KEY_BYTES = 512 +_MAX_TEXT_BYTES = 4000 +_BATCH_SIZE = 500 +_METRICS = { + "DEFAULT": "COSINE", + "cosine_distance": "COSINE", + "cosine_similarity": "COSINE", + "euclidean_distance": "EUCLIDEAN", + "euclidean_squared_distance": "EUCLIDEAN_SQUARED", + "manhattan": "MANHATTAN", + "negative_dot_prod": "DOT", + "dot_prod": "DOT", +} +_ORDER_OPERATORS = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<="} + + +class OracleSettings(TypedDict, total=False): + """Connection settings resolved from arguments, a selected .env file, or ``ORACLE_`` variables.""" + + dsn: str | None + user: str | None + password: SecretString | None + + +def _quote_identifier(name: str) -> str: + if not isinstance(name, str) or not name or "\0" in name: + raise ValueError("Oracle identifiers must be nonempty strings without NUL.") + try: + length = len(name.encode("utf-8")) + except UnicodeEncodeError as exc: + raise ValueError("Oracle identifiers must be valid UTF-8.") from exc + if length > _MAX_IDENTIFIER_BYTES: + raise ValueError("Oracle identifiers cannot exceed 128 UTF-8 bytes.") + return '"' + name.replace('"', '""') + '"' + + +def _validate_operation_options(options: Mapping[str, Any] | None) -> None: + if options: + raise NotImplementedError("Oracle does not support operation_options.") + + +def _validate_scalar(field: VectorStoreField, value: Any) -> Any: + if value is None: + if field.field_type == "key": + raise ValueError("Oracle keys cannot be null.") + return None + kind = field.type_ + if kind == "UUID": + if not isinstance(value, UUID | str): + raise TypeError(f"Field '{field.name}' requires a UUID.") + return str(UUID(str(value))) + if kind == "str": + if not isinstance(value, str): + raise TypeError(f"Field '{field.name}' requires a string.") + if not value: + raise ValueError("Oracle cannot store an empty string distinctly from NULL.") + limit = _MAX_KEY_BYTES if field.field_type == "key" else _MAX_TEXT_BYTES + if len(value.encode("utf-8")) > limit: + raise ValueError(f"Field '{field.name}' exceeds {limit} UTF-8 bytes.") + return value + if kind == "int": + if type(value) is not int: + raise TypeError(f"Field '{field.name}' requires an integer.") + if not -(2**63) <= value < 2**63: + raise ValueError(f"Field '{field.name}' exceeds the signed 64-bit integer range.") + return value + if kind == "float": + if type(value) not in (int, float): + raise TypeError(f"Field '{field.name}' requires a number.") + try: + number = float(value) + except OverflowError as exc: + raise ValueError(f"Field '{field.name}' requires a finite number.") from exc + if not math.isfinite(number): + raise ValueError(f"Field '{field.name}' requires a finite number.") + return number + if kind == "bool": + if type(value) is not bool: + raise TypeError(f"Field '{field.name}' requires a boolean.") + return int(value) + raise NotImplementedError(f"Oracle field type '{kind}' is not supported.") + + +def _prepare_vector(field: VectorStoreField, value: Any) -> array[Any] | None: + if value is None: + return None + if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)): + tolist = getattr(value, "tolist", None) + value = tolist() if callable(tolist) else value + if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)): + raise TypeError(f"Oracle vector '{field.name}' requires a dense numeric sequence.") + elements = cast(Sequence[Any], value) + if len(elements) != field.dimensions: + raise ValueError(f"Oracle vector '{field.name}' expects {field.dimensions} dimensions; got {len(elements)}.") + element_type = field.type_ or "float32" + if element_type == "int8": + if any(type(item) is not int or not -128 <= item <= 127 for item in elements): + raise ValueError(f"Oracle vector '{field.name}' requires signed int8 elements.") + return array("b", elements) + result: list[float] = [] + for item in elements: + if type(item) not in (int, float): + raise TypeError(f"Oracle vector '{field.name}' requires numeric elements, not booleans.") + try: + number = float(item) + except OverflowError as exc: + raise ValueError(f"Oracle vector '{field.name}' requires finite elements.") from exc + if not math.isfinite(number): + raise ValueError(f"Oracle vector '{field.name}' requires finite elements.") + result.append(number) + try: + prepared = array("d" if element_type == "float64" else "f", result) + except OverflowError as exc: + raise ValueError(f"Oracle vector '{field.name}' contains out-of-range elements.") from exc + if not all(math.isfinite(item) for item in prepared): + raise ValueError(f"Oracle vector '{field.name}' contains out-of-range elements.") + return prepared + + +def _column_type(field: VectorStoreField) -> str: + if field.field_type == "vector": + dimensions = field.dimensions + if type(dimensions) is not int or not 1 <= dimensions <= 65535: + raise ValueError("Oracle vector dimensions must be an integer between 1 and 65535.") + storage_type = { + None: "FLOAT32", + "float": "FLOAT32", + "float32": "FLOAT32", + "float64": "FLOAT64", + "int8": "INT8", + }[field.type_] + return f"VECTOR({dimensions}, {storage_type})" + if field.type_ is None: + raise ValueError(f"Oracle field '{field.name}' requires an explicit type.") + if field.field_type == "key": + return {"str": "VARCHAR2(512 BYTE)", "UUID": "VARCHAR2(36 BYTE)", "int": "NUMBER(19,0)"}[field.type_] + return {"str": "VARCHAR2(4000 BYTE)", "int": "NUMBER(19,0)", "float": "BINARY_DOUBLE", "bool": "NUMBER(1,0)"}[ + field.type_ + ] + + +def _bind_type(field: VectorStoreField) -> Any: + if field.field_type == "vector": + return cast(Any, oracledb.DB_TYPE_VECTOR) # pyright: ignore[reportUnknownMemberType] + if field.type_ in ("str", "UUID"): + return ( + _MAX_KEY_BYTES + if field.field_type == "key" and field.type_ == "str" + else (36 if field.type_ == "UUID" else _MAX_TEXT_BYTES) + ) + if field.type_ == "float": + return cast(Any, oracledb.DB_TYPE_BINARY_DOUBLE) # pyright: ignore[reportUnknownMemberType] + return cast(Any, oracledb.DB_TYPE_NUMBER) # pyright: ignore[reportUnknownMemberType] + + +def _oracle_error_code(exc: oracledb.DatabaseError) -> int | None: + error = exc.args[0] if exc.args else None + code = getattr(error, "code", None) + return code if isinstance(code, int) else None + + +class _Client: + """Own a lazy async pool or borrow an async pool/connection.""" + + def __init__( + self, + *, + dsn: str | None = None, + user: str | None = None, + password: SecretString | None = None, + client: OracleClient | None = None, + ) -> None: + if client is not None: + if not isinstance(client, (oracledb.AsyncConnection, oracledb.AsyncConnectionPool)): + raise TypeError("client must be an oracledb AsyncConnection or AsyncConnectionPool.") + if any(value is not None for value in (dsn, user, password)): + raise ValueError("An Oracle client cannot be combined with connection settings.") + else: + if not isinstance(dsn, str) or not dsn.strip(): + raise ValueError("A nonempty Oracle DSN is required.") + if not isinstance(user, str) or not user.strip(): + raise ValueError("A nonempty Oracle user is required.") + if not isinstance(password, SecretString) or not password.get_secret_value(): + raise ValueError("A nonempty Oracle password is required.") + self._client = client + self._dsn = dsn + self._user = user + self._password = password + self._pool: oracledb.AsyncConnectionPool | None = None + self._lock = asyncio.Lock() + self._closed = False + + def ensure_open(self) -> None: + if self._closed: + raise RuntimeError("The Oracle client is closed.") + + async def _get_client(self) -> OracleClient: + self.ensure_open() + if self._client is not None: + return self._client + async with self._lock: + self.ensure_open() + if self._pool is None: + if self._user is None or self._dsn is None or self._password is None: + raise RuntimeError("Oracle connection settings are missing.") + self._pool = oracledb.create_pool_async( + user=self._user, password=self._password.get_secret_value(), dsn=self._dsn, min=0, max=4 + ) + return self._pool + + @asynccontextmanager + async def connection(self, *, write: bool = False) -> AsyncGenerator[oracledb.AsyncConnection]: + """Commit or roll back pooled writes; a borrowed connection remains caller-transactional.""" + try: + client = await self._get_client() + if isinstance(client, oracledb.AsyncConnectionPool): + async with client.acquire() as connection: # pyright: ignore[reportUnknownMemberType] + try: + yield connection + if write: + await connection.commit() + except (Exception, asyncio.CancelledError): + if write: + await connection.rollback() + raise + else: + yield client + except oracledb.Error as exc: + raise IntegrationException("Oracle operation failed; inspect the chained driver exception.") from exc + + async def close(self) -> None: + """Close only a connector-created pool; a borrowed client remains open.""" + async with self._lock: + if not self._closed: + if self._pool is not None: + await self._pool.close() + self._closed = True + + +def _create_client( + *, + dsn: str | None, + user: str | None, + password: str | SecretString | None, + client: OracleClient | None, + env_file_path: str | None, + env_file_encoding: str | None, +) -> _Client: + if client is not None: + if any(value is not None for value in (dsn, user, password, env_file_path, env_file_encoding)): + raise ValueError("client cannot be combined with dsn, user, password, or .env settings.") + return _Client(client=client) + settings = load_settings( + OracleSettings, + env_prefix="ORACLE_", + dsn=dsn, + user=user, + password=password, + env_file_path=env_file_path, + env_file_encoding=env_file_encoding, + ) + return _Client(dsn=settings.get("dsn"), user=settings.get("user"), password=settings.get("password")) + + +def _scalar_field(definition: VectorStoreCollectionDefinition, name: str) -> VectorStoreField: + if "." in name: + raise NotImplementedError("Oracle filters do not support nested field paths.") + field = definition.try_get_field(name) + if field is None: + raise ValueError(f"Unknown Oracle field '{name}'.") + if field.field_type == "vector": + raise NotImplementedError("Oracle cannot filter or order vector columns.") + return field + + +class _FilterCompiler: + """Compile core-validated scalar filters to two-valued, bound Oracle SQL.""" + + def __init__(self, definition: VectorStoreCollectionDefinition) -> None: + self.definition = definition + self.binds: dict[str, Any] = {} + + def compile(self, expression: FilterExpression) -> tuple[str, dict[str, Any]]: + """Compile an expression and its bind values without retaining caller state.""" + self.binds = {} + return self._condition(expression), self.binds + + def _bind(self, value: Any) -> str: + name = f"f{len(self.binds)}" + self.binds[name] = value + return f":{name}" + + def _equality(self, field: VectorStoreField, value: Any) -> str: + column = _quote_identifier(field.storage_name or field.name) + if value is None: + return f"{column} IS NULL" + kind = field.type_ + if kind in ("int", "float") and type(value) in (int, float): + if type(value) is float and not math.isfinite(value): + raise ValueError("Numeric filter values must be finite.") + if kind == "int": + if type(value) is float: + if not value.is_integer() or not -(2**63) <= value < 2**63: + return "1=0" + value = int(value) + elif not -(2**63) <= value < 2**63: + return "1=0" + elif type(value) is int: + try: + numerator, denominator = float(value).as_integer_ratio() + if numerator != value * denominator: + return "1=0" + except OverflowError: + return "1=0" + adapted = value if kind == "int" else float(value) + elif ( + (kind == "bool" and type(value) is not bool) + or (kind != "bool" and isinstance(value, bool)) + or (kind == "UUID" and not isinstance(value, UUID | str)) + or (kind == "str" and not isinstance(value, str)) + or (kind in ("int", "float") and type(value) not in (int, float)) + ): + return "1=0" + else: + adapted = _validate_scalar(field, value) + return f"({column} IS NOT NULL AND {column} = {self._bind(adapted)})" + + def _condition(self, expression: FilterExpression) -> str: + if isinstance(expression, FilterGroup): + parts = [self._condition(child) for child in expression.filters] + if expression.operator == "not": + return f"(NOT ({parts[0]}))" + joiner = " AND " if expression.operator == "and" else " OR " + return f"({joiner.join(parts)})" + field = _scalar_field(self.definition, expression.field_name) + column = _quote_identifier(field.storage_name or field.name) + op, value = expression.operator, expression.value + if op == "exists": + return "1=1" + if op == "is_null": + return f"{column} IS NULL" + if op == "is_not_null": + return f"{column} IS NOT NULL" + if op in ("eq", "ne"): + equality = self._equality(field, value) + return equality if op == "eq" else f"(NOT ({equality}))" + if op in ("in", "not_in"): + parts = [self._equality(field, item) for item in cast(Collection[Any], value)] + membership = f"({' OR '.join(parts)})" if parts else "1=0" + if op == "not_in": + membership = f"(NOT ({membership}))" + return f"({column} IS NOT NULL AND {membership})" + if op in _ORDER_OPERATORS or op == "between": + if field.type_ not in ("int", "float"): + raise NotImplementedError("Oracle ordered filters support only numeric fields.") + operands = cast(Sequence[Any], value) if op == "between" else [value] + adapted: list[Any] = [] + for item in operands: + if item is None or isinstance(item, bool) or type(item) not in (int, float): + raise TypeError("Ordered filter operands must be non-null numbers, not booleans.") + if type(item) is float and not math.isfinite(item): + raise ValueError("Ordered filter numbers must be finite.") + if field.type_ == "int" and type(item) is float and not item.is_integer(): + raise NotImplementedError("Oracle integer range filters require integer operands.") + adapted.append(_validate_scalar(field, int(item) if field.type_ == "int" else item)) + comparison = ( + f"{column} BETWEEN {self._bind(adapted[0])} AND {self._bind(adapted[1])}" + if op == "between" + else f"{column} {_ORDER_OPERATORS[op]} {self._bind(adapted[0])}" + ) + return f"({column} IS NOT NULL AND {comparison})" + if op in ("starts_with", "ends_with", "contains_text"): + if field.type_ != "str": + raise TypeError("Oracle text filtering requires a string column.") + if not isinstance(value, str): + raise TypeError("Oracle text filtering requires a string operand.") + escaped = value.replace("!", "!!").replace("%", "!%").replace("_", "!_") + pattern = ("%" if op != "starts_with" else "") + escaped + ("%" if op != "ends_with" else "") + return f"({column} IS NOT NULL AND {column} LIKE {self._bind(pattern)} ESCAPE '!')" + raise NotImplementedError(f"Unsupported Oracle filter operator '{op}'.") + + +class OracleCollection( + BaseVectorCollection[KeyT, ModelT], + BaseVectorSearch[KeyT, ModelT], + Generic[KeyT, ModelT], +): + """Store typed records and search Oracle Database 23ai native VECTOR columns.""" + + supported_key_types: ClassVar[set[str] | None] = {"str", "int", "UUID"} + supported_vector_types: ClassVar[set[str] | None] = {"float", "float32", "float64", "int8"} + supported_search_types: ClassVar[set[SearchType]] = {"vector"} + + def __init__( + self, + record_type: type[ModelT], + *, + definition: VectorStoreCollectionDefinition | None = None, + collection_name: str | None = None, + embedding_generator: EmbeddingClient | None = None, + dsn: str | None = None, + user: str | None = None, + password: str | SecretString | None = None, + client: OracleClient | None = None, + env_file_path: str | None = None, + env_file_encoding: str | None = None, + _shared_client: _Client | None = None, + ) -> None: + """Initialize a collection with explicit, selected .env, or environment credentials. + + Args: + record_type: Registered application model, or dict with an explicit definition. + definition: Explicit dictionary model definition. + collection_name: Override for the model's table name in the current Oracle schema. + embedding_generator: Optional local embedding generator. + dsn: Oracle connect string; defaults to ``ORACLE_DSN``. + user: Oracle database user; defaults to ``ORACLE_USER``. + password: Database password or AF SecretString; defaults to ``ORACLE_PASSWORD``. + client: Borrowed async connection or pool in place of connection settings. + env_file_path: Optional selected .env file; no implicit discovery. + env_file_encoding: Encoding of the selected .env file. + """ + if _shared_client is not None and any( + item is not None for item in (dsn, user, password, client, env_file_path, env_file_encoding) + ): + raise ValueError("Shared Oracle clients cannot be combined with connection settings.") + super().__init__( + record_type, + definition=definition, + collection_name=collection_name, + embedding_generator=embedding_generator, + managed_client=_shared_client is None and client is None, + ) + self._table = _quote_identifier(self.collection_name) + self._fields = tuple(self.definition.fields) + self._client = _shared_client or _create_client( + dsn=dsn, + user=user, + password=password, + client=client, + env_file_path=env_file_path, + env_file_encoding=env_file_encoding, + ) + self._shared_client = _shared_client is not None + + def _validate_data_model(self) -> None: + super()._validate_data_model() + if not self.definition.vector_fields: + raise ValueError("Oracle collections require a vector field.") + for field in self.definition.fields: + _quote_identifier(field.storage_name or field.name) + if field.provider_annotations: + raise NotImplementedError("Oracle provider annotations are not supported.") + if field.is_full_text_indexed or field.is_indexed: + raise NotImplementedError("Oracle data and full-text indexes are not supported.") + if field.is_auto_generated: + raise NotImplementedError("Oracle requires application-provided keys.") + if field.field_type == "vector": + if field.index_kind not in ("default", "flat"): + raise NotImplementedError(f"Oracle index kind '{field.index_kind}' is not supported.") + if field.distance_function not in _METRICS: + raise NotImplementedError(f"Oracle distance function '{field.distance_function}' is not supported.") + elif field.field_type == "key" and field.type_ not in ("str", "int", "UUID"): + raise NotImplementedError(f"Oracle key type '{field.type_}' is not supported.") + elif field.field_type == "data" and field.type_ not in ("str", "int", "float", "bool"): + raise NotImplementedError(f"Oracle data type '{field.type_}' is not supported.") + _column_type(field) + + async def __aenter__(self) -> Self: + """Enter the collection context without connecting until first use.""" + return self + + async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: + """Close only a collection-owned pool.""" + await self.close() + + async def close(self) -> None: + """Release this collection's resources, not a borrowed or store-shared client.""" + if not self._shared_client: + await self._client.close() + + async def collection_exists(self, *, operation_options: Mapping[str, Any] | None = None) -> bool: + """Check whether the table exists in the current user's schema.""" + _validate_operation_options(operation_options) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + await cursor.execute("SELECT 1 FROM USER_TABLES WHERE TABLE_NAME = :name", {"name": self.collection_name}) + return await cursor.fetchone() is not None + + async def ensure_collection_exists(self, *, operation_options: Mapping[str, Any] | None = None) -> None: + """Create this table if absent; do not migrate existing tables or create indexes.""" + _validate_operation_options(operation_options) + columns: list[str] = [] + for field in self._fields: + column = f"{_quote_identifier(field.storage_name or field.name)} {_column_type(field)}" + columns.append(column + (" PRIMARY KEY" if field.field_type == "key" else "")) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + await cursor.execute("SELECT 1 FROM USER_TABLES WHERE TABLE_NAME = :name", {"name": self.collection_name}) + if await cursor.fetchone() is not None: + return + try: + await cursor.execute(f"CREATE TABLE {self._table} ({', '.join(columns)})") + except oracledb.DatabaseError as exc: + if _oracle_error_code(exc) != 955: + raise + await cursor.execute( + "SELECT 1 FROM USER_TABLES WHERE TABLE_NAME = :name", {"name": self.collection_name} + ) + if await cursor.fetchone() is None: + raise + + async def ensure_collection_deleted(self, *, operation_options: Mapping[str, Any] | None = None) -> None: + """Drop only this table, without modifying other database objects.""" + _validate_operation_options(operation_options) + if await self.collection_exists(): + await _drop_table(self._client, self.collection_name) + + def _deserialize_store_models_to_dicts( + self, + records: Sequence[Any], + *, + context: Mapping[str, Any] | None = None, + ) -> Sequence[dict[str, Any]]: + result = super()._deserialize_store_models_to_dicts(records, context=context) + for record in result: + for field in self._fields: + name = field.storage_name or field.name + if field.field_type == "vector" and record.get(name) is not None: + if not isinstance(record[name], array): + raise IntegrationInvalidResponseException(f"Oracle returned an invalid vector for '{name}'.") + record[name] = record[name].tolist() + return result + + def _prepare_filter(self, filter: FilterExpression | None) -> tuple[str, dict[str, Any]]: + return ("1=1", {}) if filter is None else _FilterCompiler(self.definition).compile(filter) + + def _prepare_order_by(self, order_by: Mapping[str, bool] | None) -> str: + parts: list[str] = [] + ordered_fields: set[str] = set() + for name, ascending in (order_by or {}).items(): + if type(ascending) is not bool: + raise TypeError("Oracle order directions must be booleans.") + field = _scalar_field(self.definition, name) + parts.append( + f"{_quote_identifier(field.storage_name or field.name)} {'ASC' if ascending else 'DESC'} NULLS LAST" + ) + ordered_fields.add(field.name) + if self.definition.key_name not in ordered_fields: + parts.append(f"{_quote_identifier(self.definition.key_field_storage_name)} ASC") + return ", ".join(parts) + + def _prepare_key(self, key: Any) -> Any: + return _validate_scalar(self.definition.key_field, key) + + def _decode_row(self, fields: Sequence[VectorStoreField], row: Sequence[Any]) -> dict[str, Any]: + if len(row) != len(fields): + raise IntegrationInvalidResponseException("Oracle returned an unexpected number of columns.") + result: dict[str, Any] = {} + for field, value in zip(fields, row, strict=True): + if value is not None and field.type_ == "UUID": + if not isinstance(value, UUID | str): + raise IntegrationInvalidResponseException(f"Oracle returned an invalid UUID for '{field.name}'.") + try: + value = UUID(str(value)) + except ValueError as exc: + raise IntegrationInvalidResponseException( + f"Oracle returned an invalid UUID for '{field.name}'." + ) from exc + elif value is not None and field.type_ == "int": + if (isinstance(value, Decimal) and value.is_finite() and value == int(value)) or ( + type(value) is float and math.isfinite(value) and abs(value) <= 2**53 and value.is_integer() + ): + value = int(value) + if type(value) is not int or not -(2**63) <= value < 2**63: + raise IntegrationInvalidResponseException(f"Oracle returned an invalid integer for '{field.name}'.") + elif value is not None and field.type_ == "float": + if isinstance(value, bool) or not isinstance(value, int | float | Decimal): + raise IntegrationInvalidResponseException(f"Oracle returned an invalid float for '{field.name}'.") + try: + value = float(value) + except OverflowError as exc: + raise IntegrationInvalidResponseException( + f"Oracle returned an invalid float for '{field.name}'." + ) from exc + if not math.isfinite(value): + raise IntegrationInvalidResponseException(f"Oracle returned an invalid float for '{field.name}'.") + elif value is not None and field.type_ == "bool": + if value not in (0, 1): + raise IntegrationInvalidResponseException(f"Oracle returned an invalid boolean for '{field.name}'.") + value = bool(value) + result[field.storage_name or field.name] = value + return result + + async def _inner_upsert( + self, records: Sequence[Any], *, operation_options: Mapping[str, Any] | None = None + ) -> Sequence[KeyT]: + _validate_operation_options(operation_options) + self._client.ensure_open() + binds: list[dict[str, Any]] = [] + keys: list[KeyT] = [] + key_index = self._fields.index(self.definition.key_field) + for record in records: + row: dict[str, Any] = {} + for index, field in enumerate(self._fields): + name = field.storage_name or field.name + row[f"p{index}"] = ( + _prepare_vector(field, record[name]) + if field.field_type == "vector" + else _validate_scalar(field, record[name]) + ) + key = row[f"p{key_index}"] + keys.append(cast(KeyT, UUID(key) if self.definition.key_field.type_ == "UUID" else key)) + binds.append(row) + if not binds: + return [] + key_column = _quote_identifier(self.definition.key_field_storage_name) + updates = [ + f"target.{_quote_identifier(field.storage_name or field.name)} = :p{index}" + for index, field in enumerate(self._fields) + if index != key_index + ] + columns = ", ".join(_quote_identifier(field.storage_name or field.name) for field in self._fields) + placeholders = ", ".join(f":p{index}" for index in range(len(self._fields))) + # SQL interpolates only quoted identifiers and fixed fragments; data remains bound. + statement = ( + f"MERGE INTO {self._table} target USING DUAL " # ruff: ignore[hardcoded-sql-expression] # nosec B608 + f"ON (target.{key_column} = :p{key_index}) " + f"WHEN MATCHED THEN UPDATE SET {', '.join(updates)} " + f"WHEN NOT MATCHED THEN INSERT ({columns}) VALUES ({placeholders})" + ) + async with ( + self._client.connection(write=True) as connection, + cast(_OracleCursor, connection.cursor()) as cursor, + ): + cursor.setinputsizes(**{f"p{index}": _bind_type(field) for index, field in enumerate(self._fields)}) + for start in range(0, len(binds), _BATCH_SIZE): + await cursor.executemany(statement, binds[start : start + _BATCH_SIZE]) + return keys + + async def _inner_get( + self, + *, + keys: Sequence[KeyT] | None = None, + filter: FilterExpression | None = None, + top: int = 10, + skip: int = 0, + order_by: Mapping[str, bool] | None = None, + include_vectors: bool = False, + operation_options: Mapping[str, Any] | None = None, + ) -> Sequence[Any]: + _validate_operation_options(operation_options) + self._client.ensure_open() + fields = tuple(field for field in self._fields if include_vectors or field.field_type != "vector") + columns = ", ".join(_quote_identifier(field.storage_name or field.name) for field in fields) + if keys is not None: + if order_by or skip or top != 10: + raise ValueError("Oracle key lookup cannot be combined with order_by, top, or skip.") + adapted = [self._prepare_key(key) for key in keys] + if not adapted: + return [] + found: dict[Any, dict[str, Any]] = {} + key_column = _quote_identifier(self.definition.key_field_storage_name) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + for start in range(0, len(adapted), _BATCH_SIZE): + chunk = adapted[start : start + _BATCH_SIZE] + binds = {f"k{index}": key for index, key in enumerate(chunk)} + placeholders = ", ".join(f":{name}" for name in binds) + statement = ( + f"SELECT {columns} FROM {self._table} " # ruff: ignore[hardcoded-sql-expression] # nosec B608 + f"WHERE {key_column} IN ({placeholders})" + ) + await cursor.execute(statement, binds) + for row in await cursor.fetchall(): + decoded = self._decode_row(fields, row) + found[self._prepare_key(decoded[self.definition.key_field_storage_name])] = decoded + return [found[key] for key in adapted if key in found] + where, binds = self._prepare_filter(filter) + order = self._prepare_order_by(order_by) + binds.update(top=top, skip=skip) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + statement = ( + f"SELECT {columns} FROM {self._table} " # ruff: ignore[hardcoded-sql-expression] # nosec B608 + f"WHERE {where} ORDER BY {order} " + "OFFSET :skip ROWS FETCH NEXT :top ROWS ONLY" + ) + await cursor.execute(statement, binds) + return [self._decode_row(fields, row) for row in await cursor.fetchall()] + + async def _inner_delete(self, keys: Sequence[KeyT], *, operation_options: Mapping[str, Any] | None = None) -> None: + _validate_operation_options(operation_options) + self._client.ensure_open() + adapted = [self._prepare_key(key) for key in keys] + if not adapted: + return + key_column = _quote_identifier(self.definition.key_field_storage_name) + async with ( + self._client.connection(write=True) as connection, + cast(_OracleCursor, connection.cursor()) as cursor, + ): + for start in range(0, len(adapted), _BATCH_SIZE): + chunk = adapted[start : start + _BATCH_SIZE] + binds = {f"k{index}": key for index, key in enumerate(chunk)} + placeholders = ", ".join(f":{name}" for name in binds) + statement = ( + f"DELETE FROM {self._table} " # ruff: ignore[hardcoded-sql-expression] # nosec B608 + f"WHERE {key_column} IN ({placeholders})" + ) + await cursor.execute(statement, binds) + + async def _inner_search( + self, + *, + search_type: SearchType, + filter: FilterExpression | None = None, + values: Any | None = None, + vector: Vector | None = None, + top: int = 3, + skip: int = 0, + include_vectors: bool = False, + vector_property_name: str | None = None, + additional_property_name: str | None = None, + score_threshold: float | None = None, + operation_options: Mapping[str, Any] | None = None, + ) -> SearchResults[Any]: + _validate_operation_options(operation_options) + if search_type != "vector" or additional_property_name is not None: + raise NotImplementedError("Oracle supports dense vector search only, not keyword-hybrid search.") + if vector is None: + raise NotImplementedError( + "Oracle has no server-side vectorization; supply a vector or embedding generator." + ) + self._client.ensure_open() + field = self.definition.try_get_vector_field(vector_property_name) + if field is None: + raise ValueError("Select a vector_property_name from the collection definition.") + prepared = _prepare_vector(field, vector) + column = _quote_identifier(field.storage_name or field.name) + metric = _METRICS[field.distance_function or "DEFAULT"] + distance = f"VECTOR_DISTANCE({column}, :query_vector, {metric})" + score = ( + f"(1 - {distance})" + if field.distance_function == "cosine_similarity" + else (f"(-{distance})" if field.distance_function == "dot_prod" else distance) + ) + where, binds = self._prepare_filter(filter) + binds["query_vector"] = prepared + predicate = f"({where}) AND {column} IS NOT NULL" + if score_threshold is not None: + if type(score_threshold) not in (int, float): + raise ValueError("Oracle score_threshold must be a finite number.") + try: + score_threshold = float(score_threshold) + except OverflowError as exc: + raise ValueError("Oracle score_threshold must be a finite number.") from exc + if not math.isfinite(score_threshold): + raise ValueError("Oracle score_threshold must be a finite number.") + cutoff = ( + 1 - score_threshold + if field.distance_function == "cosine_similarity" + else -score_threshold + if field.distance_function == "dot_prod" + else score_threshold + ) + binds["threshold"] = cutoff + predicate += f" AND {distance} <= :threshold" + binds.update(top=top, skip=skip) + fields = tuple(item for item in self._fields if include_vectors or item.field_type != "vector") + columns = ", ".join(_quote_identifier(item.storage_name or item.name) for item in fields) + statement = ( + f"SELECT {columns}, {score} FROM {self._table} " # ruff: ignore[hardcoded-sql-expression] # nosec B608 + f"WHERE {predicate} " + f"ORDER BY {distance} ASC, {_quote_identifier(self.definition.key_field_storage_name)} ASC " + "OFFSET :skip ROWS FETCH NEXT :top ROWS ONLY" + ) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + cursor.setinputsizes(query_vector=_bind_type(field)) + await cursor.execute(statement, binds) + rows = await cursor.fetchall() + results: list[dict[str, Any]] = [] + for row in rows: + record = self._decode_row(fields, row[:-1]) + if row[-1] is None: + raise IntegrationInvalidResponseException("Oracle returned a null vector distance.") + score_value = float(row[-1]) + if not math.isfinite(score_value): + raise IntegrationInvalidResponseException("Oracle returned a nonfinite vector distance.") + results.append({"record": record, "score": score_value}) + return SearchResults(results, metadata={"distance_function": field.distance_function or "DEFAULT"}) + + def _get_record_from_result(self, result: Any) -> Any: + return result["record"] + + def _get_score_from_result(self, result: Any) -> float | None: + return result["score"] + + +async def _drop_table(client: _Client, name: str) -> None: + async with client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + try: + await cursor.execute(f"DROP TABLE {_quote_identifier(name)} PURGE") + except oracledb.DatabaseError as exc: + if _oracle_error_code(exc) != 942: + raise + + +class OracleStore(BaseVectorStore): + """Create Oracle vector collections sharing one async pool or connection.""" + + def __init__( + self, + *, + dsn: str | None = None, + user: str | None = None, + password: str | SecretString | None = None, + client: OracleClient | None = None, + embedding_generator: EmbeddingClient | None = None, + env_file_path: str | None = None, + env_file_encoding: str | None = None, + ) -> None: + """Initialize a store without opening its connector-owned pool. + + Args: + dsn: Oracle connect string; defaults to ``ORACLE_DSN``. + user: Oracle database user; defaults to ``ORACLE_USER``. + password: Database password or AF SecretString; defaults to ``ORACLE_PASSWORD``. + client: Borrowed async connection or pool in place of connection settings. + embedding_generator: Default local embedding client for child collections. + env_file_path: Optional selected .env file; no implicit discovery. + env_file_encoding: Encoding of the selected .env file. + """ + super().__init__(embedding_generator=embedding_generator, managed_client=client is None) + self._client = _create_client( + dsn=dsn, + user=user, + password=password, + client=client, + env_file_path=env_file_path, + env_file_encoding=env_file_encoding, + ) + + def get_collection( + self, + record_type: type[ModelT], + *, + definition: VectorStoreCollectionDefinition | None = None, + collection_name: str | None = None, + embedding_generator: EmbeddingClient | None = None, + ) -> OracleCollection[Any, ModelT]: + """Create a collection borrowing this store's connection lifecycle.""" + return OracleCollection( + record_type, + definition=definition, + collection_name=collection_name, + embedding_generator=embedding_generator if embedding_generator is not None else self.embedding_generator, + _shared_client=self._client, + ) + + async def list_collection_names(self, *, operation_options: Mapping[str, Any] | None = None) -> Sequence[str]: + """List base table names in the current Oracle user schema.""" + _validate_operation_options(operation_options) + async with self._client.connection() as connection, cast(_OracleCursor, connection.cursor()) as cursor: + await cursor.execute("SELECT TABLE_NAME FROM USER_TABLES ORDER BY TABLE_NAME") + names: list[str] = [] + for row in await cursor.fetchall(): + if len(row) != 1 or not isinstance(row[0], str): + raise IntegrationInvalidResponseException("Oracle returned an invalid table name.") + names.append(row[0]) + return names + + async def _inner_ensure_collection_deleted( + self, collection_name: str, *, operation_options: Mapping[str, Any] | None = None + ) -> None: + _validate_operation_options(operation_options) + await _drop_table(self._client, collection_name) + + async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: + """Close the owned pool when exiting the store context.""" + await self.close() + + async def close(self) -> None: + """Close the store, leaving injected clients untouched.""" + await self._client.close() diff --git a/python/packages/oracle/agent_framework_oracle/py.typed b/python/packages/oracle/agent_framework_oracle/py.typed new file mode 100644 index 00000000000..e69de29bb2d diff --git a/python/packages/oracle/pyproject.toml b/python/packages/oracle/pyproject.toml new file mode 100644 index 00000000000..278116128f3 --- /dev/null +++ b/python/packages/oracle/pyproject.toml @@ -0,0 +1,63 @@ +[project] +name = "agent-framework-oracle" +description = "Oracle Database native vector stores for Microsoft Agent Framework." +authors = [{ name = "Microsoft", email = "af-support@microsoft.com" }] +readme = "README.md" +requires-python = ">=3.10" +version = "1.0.0a260923" +license-files = ["LICENSE"] +urls.homepage = "https://aka.ms/agent-framework" +urls.source = "https://github.com/microsoft/agent-framework/tree/main/python" +urls.issues = "https://github.com/microsoft/agent-framework/issues" +classifiers = [ + "License :: OSI Approved :: MIT License", + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Typing :: Typed", +] +dependencies = [ + "agent-framework-core>=1.19.0,<2", + "oracledb>=2.2.0,<4", +] + +[tool.ruff] +extend = "../../pyproject.toml" + +[tool.ruff.lint.per-file-ignores] +"samples/**" = ["D", "INP", "ERA001", "RUF", "S", "T201", "CPY"] +"tests/**" = ["D", "INP", "TD", "ERA001", "RUF", "S"] + +[tool.pyright] +extends = "../../pyproject.toml" +include = ["agent_framework_oracle"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra -q -r fEX" +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +timeout = 120 +markers = ["integration: requires an explicitly designated disposable Oracle Database 23ai+ schema"] + +[tool.coverage.run] +omit = ["**/__init__.py"] + +[tool.poe] +executor.type = "uv" +include = "../../shared_tasks.toml" + +[tool.poe.tasks.test] +cmd = 'pytest -m "not integration" --cov=agent_framework_oracle --cov-report=term-missing:skip-covered tests' + +[tool.poe.tasks.test-integration] +cmd = "pytest -m integration tests" + +[build-system] +requires = ["flit-core>=3.11,<4.0"] +build-backend = "flit_core.buildapi" diff --git a/python/packages/oracle/samples/oracle_vectors.py b/python/packages/oracle/samples/oracle_vectors.py new file mode 100644 index 00000000000..11769316678 --- /dev/null +++ b/python/packages/oracle/samples/oracle_vectors.py @@ -0,0 +1,67 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from typing import Annotated +from uuid import uuid4 + +from agent_framework import Filter, VectorStoreField, vectorstoremodel + +from agent_framework_oracle import OracleStore + +""" +Use Oracle Database 23ai native vectors with local precomputed embeddings. + +Environment variables: + ORACLE_DSN - Oracle connect string for a disposable 23ai+ schema + ORACLE_USER - User permitted to create and drop tables in that schema + ORACLE_PASSWORD - Password for that user + +Run from python/: + uv run --package agent-framework-oracle python packages/oracle/samples/oracle_vectors.py + +The sample creates and drops only its uniquely named table. +""" + + +@vectorstoremodel +@dataclass +class Note: + id: Annotated[str, VectorStoreField("key")] + text: Annotated[str, VectorStoreField("data", storage_name="body")] + category: Annotated[str, VectorStoreField("data")] + embedding: Annotated[list[float] | None, VectorStoreField("vector", dimensions=3)] = None + + +async def main() -> None: + async with OracleStore() as store: + collection = store.get_collection(Note, collection_name=f"AF_NOTE_{uuid4().hex}") + await collection.ensure_collection_exists() + try: + await collection.upsert( + [ + Note("oracle", "Oracle supports native VECTOR columns", "database", [1, 0, 0]), + Note("travel", "A travel journal", "travel", [0, 1, 0]), + ], + generate_vectors=False, + ) + results = await collection.search( + vector=[1, 0, 0], + filter=Filter("category", "eq", "database"), + score_threshold=0.1, + ) + async for result in results: + print(result["record"].text, result["score"]) + print((await collection.get(["oracle"], include_vectors=True))[0].embedding) + finally: + await collection.ensure_collection_deleted() + + +if __name__ == "__main__": + asyncio.run(main()) + +# Expected output: +# Oracle supports native VECTOR columns 0.0 +# [1.0, 0.0, 0.0] diff --git a/python/packages/oracle/tests/oracle/conftest.py b/python/packages/oracle/tests/oracle/conftest.py new file mode 100644 index 00000000000..ac94102ca12 --- /dev/null +++ b/python/packages/oracle/tests/oracle/conftest.py @@ -0,0 +1,112 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from agent_framework import VectorStoreCollectionDefinition, VectorStoreField + +from agent_framework_oracle import OracleCollection + + +@pytest.fixture +def definition_factory(): + def make( + *, + key_type="str", + generated=False, + vector_type="float32", + dimensions=3, + index_kind="flat", + distance_function="DEFAULT", + second_vector=False, + data_type="str", + ): + fields = [ + VectorStoreField("key", name="id", type_=key_type, storage_name='doc"id', is_auto_generated=generated), + VectorStoreField("data", name="text", type_=data_type, storage_name="body text"), + VectorStoreField("data", name="number", type_="int"), + VectorStoreField("data", name="flag", type_="bool"), + VectorStoreField( + "vector", + name="embedding", + type_=vector_type, + storage_name="dense vector", + dimensions=dimensions, + index_kind=index_kind, + distance_function=distance_function, + ), + ] + if second_vector: + fields.append( + VectorStoreField( + "vector", + name="second", + type_="float64", + dimensions=dimensions, + ) + ) + return VectorStoreCollectionDefinition(fields, collection_name="AF_DOCUMENTS") + + return make + + +@pytest.fixture +def record_factory(): + def make(id="one", *, text="Hello Oracle", number=2, flag=True, embedding=None, second=None): + record = { + "id": id, + "text": text, + "number": number, + "flag": flag, + "embedding": embedding if embedding is not None else [1.0, 0.0, 0.0], + } + if second is not None: + record["second"] = second + return record + + return make + + +@pytest.fixture +def collection(definition_factory): + return OracleCollection( + dict, + definition=definition_factory(), + dsn="unused", + user="example", + password="not-used", + ) + + +class FakeCursor: + def __init__(self): + self.execute = AsyncMock() + self.executemany = AsyncMock() + self.fetchone = AsyncMock(return_value=None) + self.fetchall = AsyncMock(return_value=[]) + self.setinputsizes = MagicMock() + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_value, traceback): + return None + + +@pytest.fixture +def mock_database(collection): + cursor = FakeCursor() + connection = MagicMock() + connection.cursor.return_value = cursor + writes = [] + + @asynccontextmanager + async def acquire(*, write=False): + writes.append(write) + yield connection + + with patch.object(collection._client, "connection", side_effect=acquire) as mocked: + yield connection, cursor, writes, mocked diff --git a/python/packages/oracle/tests/oracle/test_collection.py b/python/packages/oracle/tests/oracle/test_collection.py new file mode 100644 index 00000000000..30461549fda --- /dev/null +++ b/python/packages/oracle/tests/oracle/test_collection.py @@ -0,0 +1,449 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +from array import array +from dataclasses import dataclass +from decimal import Decimal +from types import SimpleNamespace +from typing import Annotated +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import UUID + +import oracledb +import pytest +from agent_framework import ( + Embedding, + Filter, + GeneratedEmbeddings, + VectorStoreCollectionDefinition, + VectorStoreField, + vectorstoremodel, +) +from agent_framework.exceptions import IntegrationException, IntegrationInvalidResponseException + +from agent_framework_oracle import OracleCollection, OracleStore +from agent_framework_oracle._vector_store import _prepare_vector + + +@pytest.mark.parametrize( + "vector_type,kind,value", + [ + ("float", "f", [0.1, 1, 0]), + ("float32", "f", [0.1, 1, 0]), + ("float64", "d", [0.1, 1, 0]), + ("int8", "b", [-128, 0, 127]), + ], +) +def test_native_vector_adapter(vector_type, kind, value, definition_factory): + field = definition_factory(vector_type=vector_type).vector_fields[0] + prepared = _prepare_vector(field, value) + assert isinstance(prepared, array) and prepared.typecode == kind + assert list(prepared) == pytest.approx(value) + + +@pytest.mark.parametrize( + "vector,error", + [ + (b"\x00\x01\x02", TypeError), + ("0,1,2", TypeError), + ([True, 0, 1], TypeError), + ([1, 2], ValueError), + ([float("nan"), 0, 1], ValueError), + ([float("inf"), 0, 1], ValueError), + ([1e100, 0, 1], ValueError), + ], +) +def test_bad_vectors_rejected(vector, error, definition_factory): + with pytest.raises(error): + _prepare_vector(definition_factory().vector_fields[0], vector) + + +@pytest.mark.parametrize("vector", [[128, 0, 1], [-129, 0, 1], [1.0, 0, 1], [True, 0, 1]]) +def test_int8_vector_is_strict(vector, definition_factory): + with pytest.raises(ValueError, match="int8"): + _prepare_vector(definition_factory(vector_type="int8").vector_fields[0], vector) + + +@pytest.mark.parametrize( + "definition", + [ + {"generated": True}, + {"vector_type": "float16"}, + {"vector_type": "int"}, + {"index_kind": "hnsw"}, + {"distance_function": "hamming"}, + {"dimensions": 65536}, + {"data_type": "dict"}, + ], +) +def test_unsupported_model_capabilities_rejected_before_io(definition, definition_factory): + with patch.object(oracledb, "create_pool_async") as create, pytest.raises((ValueError, NotImplementedError)): + OracleCollection(dict, definition=definition_factory(**definition), dsn="unused", user="user", password="pass") + create.assert_not_called() + + +def test_missing_vector_and_data_index_rejected_before_connecting(): + with pytest.raises(ValueError, match="vector"): + OracleCollection( + dict, + definition=VectorStoreCollectionDefinition([VectorStoreField("key", name="id", type_="str")]), + collection_name="AF_ONLY_KEY", + dsn="unused", + user="user", + password="pass", + ) + with pytest.raises(NotImplementedError, match="indexes"): + OracleCollection( + dict, + definition=VectorStoreCollectionDefinition([ + VectorStoreField("key", name="id", type_="str"), + VectorStoreField("data", name="text", type_="str", is_indexed=True), + VectorStoreField("vector", name="vector", dimensions=3), + ]), + collection_name="AF_INDEXED", + dsn="unused", + user="user", + password="pass", + ) + + +async def test_lifecycle_sql_is_scoped_and_identifiers_are_quoted(collection, mock_database): + _, cursor, writes, _ = mock_database + cursor.fetchone.side_effect = [None, ("exists",), ("exists",)] + await collection.ensure_collection_exists() + sql = cursor.execute.call_args_list[1].args[0] + assert sql.startswith('CREATE TABLE "AF_DOCUMENTS" (') + assert '"doc""id" VARCHAR2(512 BYTE) PRIMARY KEY' in sql + assert '"dense vector" VECTOR(3, FLOAT32)' in sql + assert all("ALTER" not in call.args[0] and "SCHEMA" not in call.args[0] for call in cursor.execute.call_args_list) + assert await collection.collection_exists() + await collection.ensure_collection_deleted() + assert cursor.execute.call_args.args[0] == 'DROP TABLE "AF_DOCUMENTS" PURGE' + assert writes == [False, False, False, False] + + +async def test_hostile_table_name_is_quoted_and_bound_in_metadata_lookup(definition_factory, mock_database): + _, cursor, _, acquire = mock_database + name = 'x"; DROP TABLE USERS; --' + collection = OracleCollection( + dict, + definition=definition_factory(), + collection_name=name, + dsn="unused", + user="user", + password="pass", + ) + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + await collection.ensure_collection_exists() + assert cursor.execute.call_args_list[0].args == ( + "SELECT 1 FROM USER_TABLES WHERE TABLE_NAME = :name", + {"name": name}, + ) + statement = cursor.execute.call_args_list[1].args[0] + assert statement.startswith('CREATE TABLE "x""; DROP TABLE USERS; --" (') + + +async def test_concurrent_create_checks_the_conflicting_object_is_a_table(collection, mock_database): + _, cursor, _, _ = mock_database + cursor.fetchone.side_effect = [None, ("exists",)] + cursor.execute.side_effect = [None, oracledb.DatabaseError(SimpleNamespace(code=955)), None] + await collection.ensure_collection_exists() + assert cursor.execute.await_count == 3 + + +async def test_upsert_merges_batch_with_typed_binds(collection, mock_database, record_factory): + _, cursor, writes, _ = mock_database + records = [record_factory("one"), record_factory("two", text="second", embedding=[0, 1, 0])] + assert await collection.upsert(records, generate_vectors=False) == ["one", "two"] + statement, params = cursor.executemany.call_args.args + assert statement.startswith('MERGE INTO "AF_DOCUMENTS" target USING DUAL') + assert 'target."doc""id" = :p0' in statement + assert 'target."body text" = :p1' in statement + assert "WHEN NOT MATCHED THEN INSERT" in statement + assert len(params) == 2 + assert params[0]["p0"] == "one" and params[1]["p1"] == "second" + assert params[0]["p3"] == 1 and params[1]["p3"] == 1 + assert isinstance(params[0]["p4"], array) and params[0]["p4"].typecode == "f" + cursor.setinputsizes.assert_called_once_with( + p0=512, + p1=4000, + p2=oracledb.DB_TYPE_NUMBER, + p3=oracledb.DB_TYPE_NUMBER, + p4=oracledb.DB_TYPE_VECTOR, + ) + assert writes == [True] + + +async def test_nullable_vectors_and_float_data_have_explicit_bind_types( + definition_factory, mock_database, record_factory +): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(data_type="float"), + dsn="unused", + user="user", + password="pass", + ) + empty_vector = record_factory("first", text=1.5) + empty_vector["embedding"] = None + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + await collection.upsert( + [empty_vector, record_factory("second", text=2.5)], + generate_vectors=False, + ) + assert cursor.executemany.call_args.args[1][0]["p4"] is None + assert cursor.executemany.call_args.args[1][1]["p4"].typecode == "f" + assert cursor.setinputsizes.call_args.kwargs["p1"] == oracledb.DB_TYPE_BINARY_DOUBLE + assert cursor.setinputsizes.call_args.kwargs["p4"] == oracledb.DB_TYPE_VECTOR + + +async def test_embedding_generation_uses_the_shared_base_contract(collection, mock_database, record_factory): + _, cursor, _, _ = mock_database + generator = MagicMock() + generator.get_embeddings = AsyncMock(return_value=GeneratedEmbeddings([Embedding(vector=[0.0, 1.0, 0.0])])) + collection.embedding_generator = generator + await collection.upsert([record_factory(embedding="source text")]) + assert list(cursor.executemany.call_args.args[1][0]["p4"]) == [0, 1, 0] + generator.get_embeddings.assert_awaited_once() + + +async def test_upsert_multiple_vectors_and_repeated_keys(definition_factory, record_factory, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(second_vector=True), + dsn="unused", + user="user", + password="pass", + ) + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + keys = await collection.upsert( + [record_factory("same", second=[0, 1, 0]), record_factory("same", second=[1, 0, 0])], + generate_vectors=False, + ) + assert keys == ["same", "same"] + assert cursor.executemany.call_args.args[1][0]["p5"].typecode == "d" + + +@pytest.mark.parametrize("bad_key", [None, "", True, 3]) +async def test_invalid_keys_rejected_without_io(collection, mock_database, record_factory, bad_key): + _, cursor, _, acquire = mock_database + with pytest.raises((TypeError, ValueError)): + await collection.upsert([record_factory(bad_key)], generate_vectors=False) + with pytest.raises((TypeError, ValueError)): + await collection.get([bad_key]) + acquire.assert_not_called() + cursor.executemany.assert_not_awaited() + + +async def test_core_checks_batch_dimensions_before_writing(collection, mock_database, record_factory): + _, _, _, acquire = mock_database + with pytest.raises(ValueError, match="index 1"): + await collection.upsert( + [record_factory("good"), record_factory("bad", embedding=[1, 0])], + generate_vectors=False, + ) + with pytest.raises(ValueError, match="dimensions"): + await collection.search(vector=[1]) + acquire.assert_not_called() + + +async def test_retrieval_projects_vectors_only_when_requested(collection, mock_database, record_factory): + _, cursor, _, _ = mock_database + cursor.fetchall.return_value = [("one", "Hello Oracle", 2, 1)] + assert await collection.get(filter=Filter("number", "gte", 2), top=1, skip=1) == [ + {k: v for k, v in record_factory().items() if k != "embedding"} + ] + statement, binds = cursor.execute.call_args.args + assert "dense vector" not in statement + assert "ORDER BY" in statement and "OFFSET :skip ROWS FETCH NEXT :top ROWS ONLY" in statement + assert binds == {"f0": 2, "top": 1, "skip": 1} + cursor.fetchall.return_value = [("one", "Hello Oracle", 2, 1, array("f", [1, 0, 0]))] + assert await collection.get(["one"], include_vectors=True) == [record_factory()] + assert '"dense vector"' in cursor.execute.call_args.args[0] + + +async def test_key_lookup_preserves_order_duplicates_and_chunks(collection, mock_database): + _, cursor, _, _ = mock_database + cursor.fetchall.side_effect = [ + [("two", "other", 2, 0), ("one", "hello", 1, 1)], + [("last", "last", 3, 1)], + ] + keys = ["one", "two", *["one"] * 498, "last"] + records = await collection.get(keys) + assert [record["id"] for record in records] == keys + assert cursor.execute.await_count == 2 + assert "IN (" in cursor.execute.call_args_list[0].args[0] + assert "IN (:k0)" in cursor.execute.call_args_list[1].args[0] + + +async def test_empty_batches_and_unsupported_options_do_not_hit_database(collection, mock_database): + _, _, _, acquire = mock_database + assert await collection.upsert([], generate_vectors=False) == [] + assert await collection.get([]) == [] + await collection.delete([]) + acquire.assert_not_called() + with pytest.raises(NotImplementedError): + await collection.get(operation_options={"sql": "DROP TABLE AF_DOCUMENTS"}) + with pytest.raises(ValueError, match="key lookup"): + await collection.get(["one"], top=1) + acquire.assert_not_called() + + +async def test_delete_is_bound_and_chunked(collection, mock_database): + _, cursor, writes, _ = mock_database + await collection.delete(["first", "second"] * 251) + assert cursor.execute.await_count == 2 + statement, binds = cursor.execute.call_args_list[0].args + assert statement.startswith('DELETE FROM "AF_DOCUMENTS" WHERE "doc""id" IN (') + assert "first" not in statement and binds["k0"] == "first" + assert writes == [True] + + +async def test_search_filter_threshold_and_paging_execute_in_database(collection, mock_database): + _, cursor, _, _ = mock_database + cursor.fetchall.return_value = [("one", "Hello Oracle", 2, 1, 0.05)] + result = await collection.search( + vector=[1, 0, 0], + filter=Filter("text", "contains_text", "%_"), + score_threshold=0.1, + top=1, + skip=2, + ) + assert result.metadata == {"distance_function": "DEFAULT"} + assert [item async for item in result] == [ + { + "record": {"id": "one", "text": "Hello Oracle", "number": 2, "flag": True}, + "score": 0.05, + } + ] + statement, binds = cursor.execute.call_args.args + assert statement.count("VECTOR_DISTANCE(") == 3 + assert statement.index("<= :threshold") < statement.index("ORDER BY") < statement.index("OFFSET :skip") + assert "COSINE" in statement and "embedding" not in statement + assert binds["f0"] == "%!%!_%" and binds["threshold"] == 0.1 + assert binds["top"] == 1 and binds["skip"] == 2 + assert isinstance(binds["query_vector"], array) + cursor.setinputsizes.assert_called_once_with(query_vector=oracledb.DB_TYPE_VECTOR) + + +@pytest.mark.parametrize( + "metric,cutoff,score_expression", + [ + ("cosine_similarity", 0.9, "1 - VECTOR_DISTANCE"), + ("dot_prod", -0.1, "-VECTOR_DISTANCE"), + ("euclidean_distance", 0.1, "VECTOR_DISTANCE"), + ("negative_dot_prod", 0.1, "VECTOR_DISTANCE"), + ], +) +async def test_metric_score_and_threshold_units(definition_factory, mock_database, metric, cutoff, score_expression): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(distance_function=metric), + dsn="unused", + user="user", + password="pass", + ) + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + await collection.search(vector=[1, 0, 0], score_threshold=0.1) + statement, binds = cursor.execute.call_args.args + assert score_expression in statement + assert binds["threshold"] == pytest.approx(cutoff) + + +async def test_invalid_search_modes_and_bad_thresholds_fail_without_io(collection, mock_database): + _, _, _, acquire = mock_database + with pytest.raises(NotImplementedError): + await collection.search("text", search_type="keyword_hybrid", vector=[1, 0, 0]) + with pytest.raises(NotImplementedError): + await collection.search("text") + with pytest.raises(NotImplementedError): + await collection.search(vector=[1, 0, 0], additional_property_name="text") + with pytest.raises(NotImplementedError): + await collection.search(vector=[1, 0, 0], operation_options={"exact": True}) + with pytest.raises(ValueError): + await collection.search(vector=[1, 0, 0], score_threshold=float("nan")) + with pytest.raises(ValueError): + await collection.search(vector=[1, 0, 0], score_threshold=10**1000) + acquire.assert_not_called() + + +async def test_invalid_search_response_is_reported(collection, mock_database): + _, cursor, _, _ = mock_database + cursor.fetchall.return_value = [("one", "text", 1, 1, None)] + with pytest.raises(IntegrationInvalidResponseException, match="null vector distance"): + await collection.search(vector=[1, 0, 0]) + cursor.fetchall.return_value = [("one",)] + with pytest.raises(IntegrationInvalidResponseException, match="number of columns"): + await collection.get() + cursor.fetchall.return_value = [("one", "text", 1, 1, float("nan"))] + with pytest.raises(IntegrationInvalidResponseException, match="nonfinite"): + await collection.search(vector=[1, 0, 0]) + cursor.fetchall.return_value = [("one", "text", float(2**63), 1)] + with pytest.raises(IntegrationInvalidResponseException, match="invalid integer"): + await collection.get() + cursor.fetchall.return_value = [("one", "text", Decimal("9223372036854775807"), 1)] + assert (await collection.get())[0]["number"] == 2**63 - 1 + + +@vectorstoremodel(collection_name="AF_TYPED") +@dataclass +class TypedDocument: + id: Annotated[UUID, VectorStoreField("key")] + text: Annotated[str, VectorStoreField("data")] + embedding: Annotated[list[float] | None, VectorStoreField("vector", dimensions=3)] = None + + +def test_typed_model_deserializes_native_vectors(): + collection = OracleCollection(TypedDocument, dsn="unused", user="user", password="pass") + record = collection.deserialize({"id": str(UUID(int=1)), "text": "one", "embedding": array("f", [1, 0, 0])}) + assert isinstance(record, TypedDocument) + assert record.id == UUID(int=1) + assert record.embedding == [1, 0, 0] + + +async def test_uuid_key_lookup_uses_normalized_bind(definition_factory, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(key_type="UUID"), + dsn="unused", + user="user", + password="pass", + ) + key = UUID(int=1) + cursor.fetchall.return_value = [(str(key), "text", 1, 1)] + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + records = await collection.get([key]) + assert [record["id"] for record in records] == [key] + assert cursor.execute.call_args.args[1] == {"k0": str(key)} + + +async def test_uuid_upsert_returns_typed_key(definition_factory, mock_database, record_factory): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(key_type="UUID"), + dsn="unused", + user="user", + password="pass", + ) + key = UUID(int=1) + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + keys = await collection.upsert([record_factory(key)], generate_vectors=False) + assert keys == [key] + assert cursor.executemany.call_args.args[1][0]["p0"] == str(key) + + +async def test_store_close_invalidates_children_but_borrowed_client_stays_open(definition_factory): + connection = MagicMock(spec=oracledb.AsyncConnection) + store = OracleStore(client=connection) + child = store.get_collection(dict, definition=definition_factory()) + await child.close() + await store.close() + connection.close.assert_not_called() + with pytest.raises(IntegrationException, match="closed"): + await child.get() diff --git a/python/packages/oracle/tests/oracle/test_filters.py b/python/packages/oracle/tests/oracle/test_filters.py new file mode 100644 index 00000000000..bdfe96da74a --- /dev/null +++ b/python/packages/oracle/tests/oracle/test_filters.py @@ -0,0 +1,130 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +import pytest +from agent_framework import Filter, FilterGroup + +from agent_framework_oracle._vector_store import _FilterCompiler, _quote_identifier + + +@pytest.mark.parametrize("name", ["", "a\0b", "a" * 129, "\u00e9" * 65]) +def test_invalid_identifier_fails(name): + with pytest.raises(ValueError): + _quote_identifier(name) + + +def test_identifiers_quote_each_name_and_escape_double_quotes(): + assert _quote_identifier('t"; DROP TABLE users; --') == '"t""; DROP TABLE users; --"' + assert _quote_identifier("SCHEMA.TABLE") == '"SCHEMA.TABLE"' + assert _quote_identifier("\u00e9" * 64) == '"' + "\u00e9" * 64 + '"' + + +def test_text_patterns_escape_literals_and_bind_values(collection): + value = "%_!'; DROP TABLE AF_DOCUMENTS; --" + sql, binds = collection._prepare_filter(Filter("text", "contains_text", value)) + assert value not in sql and "DROP TABLE" not in sql + assert '"body text" LIKE :f0 ESCAPE' in sql + assert binds == {"f0": "%!%!_!!'; DROP TABLE AF!_DOCUMENTS; --%"} + + +def test_oracle_empty_string_is_rejected_instead_of_becoming_null(collection): + with pytest.raises(ValueError, match="empty string"): + collection._prepare_filter(Filter("text", "eq", "")) + + +@pytest.mark.parametrize( + "operator,expected", + [ + ("is_null", '"number" IS NULL'), + ("is_not_null", '"number" IS NOT NULL'), + ("exists", "1=1"), + ], +) +def test_null_and_exists_predicates(collection, operator, expected): + condition, binds = collection._prepare_filter(Filter("number", operator)) + assert expected in condition + assert binds == {} + + +@pytest.mark.parametrize("value", [True, "1", [], {}]) +def test_numeric_equality_does_not_coerce_invalid_values(collection, value): + sql, binds = collection._prepare_filter(Filter("number", "eq", value)) + assert sql == "1=0" and binds == {} + sql, binds = collection._prepare_filter(Filter("number", "ne", value)) + assert sql == "(NOT (1=0))" and binds == {} + + +def test_boolean_equality_does_not_coerce_numbers(collection): + sql, binds = collection._prepare_filter(Filter("flag", "eq", True)) + assert '"flag" = :f0' in sql and binds == {"f0": 1} + sql, binds = collection._prepare_filter(Filter("flag", "eq", 1)) + assert sql == "1=0" and binds == {} + + +@pytest.mark.parametrize( + "operator,value,expected", + [ + ("in", [], "1=0"), + ("not_in", [], "(NOT (1=0))"), + ("in", [True, 2, None], "IS NOT NULL"), + ("not_in", [True, 2, None], "IS NOT NULL"), + ], +) +def test_membership_has_portable_null_and_empty_semantics(collection, operator, value, expected): + sql, binds = collection._prepare_filter(Filter("number", operator, value)) + assert expected in sql + assert list(binds.values()) == ([2] if value else []) + + +def test_groups_preserve_parameter_order_and_do_not_leak_state(definition_factory): + compiler = _FilterCompiler(definition_factory()) + sql, binds = compiler.compile( + FilterGroup( + "and", + [ + Filter("id", "eq", "first"), + FilterGroup("or", [Filter("number", "between", (2, 3)), Filter("text", "eq", "last")]), + ], + ) + ) + assert sql.index('"doc""id"') < sql.index('"number"') < sql.index('"body text"') + assert binds == {"f0": "first", "f1": 2, "f2": 3, "f3": "last"} + _, next_binds = compiler.compile(Filter("id", "eq", "new")) + assert next_binds == {"f0": "new"} + assert binds["f0"] == "first" + + +@pytest.mark.parametrize( + "expression,error", + [ + (Filter("text.name", "eq", "x"), NotImplementedError), + (Filter("unknown", "eq", "x"), ValueError), + (Filter("embedding", "is_null"), NotImplementedError), + (Filter("text", "oracle.raw", "x"), NotImplementedError), + (Filter("text", "contains", "x"), NotImplementedError), + (Filter("text", "gt", "x"), NotImplementedError), + (Filter("number", "gt", True), TypeError), + (Filter("number", "between", (1, None)), TypeError), + (Filter("number", "gt", 1.5), NotImplementedError), + ], +) +def test_unsupported_filters_fail_explicitly(collection, expression, error): + with pytest.raises(error): + collection._prepare_filter(expression) + + +def test_ordering_is_quoted_and_direction_is_checked(collection): + sql = collection._prepare_order_by({"text": False}) + assert sql == '"body text" DESC NULLS LAST, "doc""id" ASC' + with pytest.raises(TypeError): + collection._prepare_order_by({"text": "DESC; DROP TABLE X"}) + with pytest.raises(NotImplementedError): + collection._prepare_order_by({"embedding": True}) + + +def test_filter_values_are_snapshot_before_translation(collection, mock_database): + expression = Filter("text", "eq", "original") + compiled = collection._prepare_filter(expression) + expression.value = "changed" + assert compiled[1] == {"f0": "original"} diff --git a/python/packages/oracle/tests/oracle/test_integration.py b/python/packages/oracle/tests/oracle/test_integration.py new file mode 100644 index 00000000000..0ec4ee86c75 --- /dev/null +++ b/python/packages/oracle/tests/oracle/test_integration.py @@ -0,0 +1,104 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +import os +from uuid import uuid4 + +import pytest +from agent_framework import Filter, VectorStoreCollectionDefinition, VectorStoreField + +from agent_framework_oracle import OracleStore + + +def _oracle_test_database_configured() -> bool: + return all(os.getenv(f"ORACLE_TEST_{name}") for name in ("DSN", "USER", "PASSWORD")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.flaky, + pytest.mark.skipif( + not _oracle_test_database_configured(), + reason="Set ORACLE_TEST_DSN, ORACLE_TEST_USER, and ORACLE_TEST_PASSWORD for a disposable Oracle 23ai+ schema.", + ), +] + + +async def test_oracle_vector_lifecycle_crud_filters_and_search(): + definition = VectorStoreCollectionDefinition([ + VectorStoreField("key", name="id", type_="str", storage_name="DOC_ID"), + VectorStoreField("data", name="text", type_="str", storage_name="BODY"), + VectorStoreField("data", name="amount", type_="int"), + VectorStoreField("data", name="flag", type_="bool"), + VectorStoreField("vector", name="embedding", type_="float32", dimensions=3), + VectorStoreField("vector", name="second", type_="float64", dimensions=3, distance_function="dot_prod"), + ]) + table = f"AF_ORACLE_TEST_{uuid4().hex}" + async with OracleStore( + dsn=os.environ["ORACLE_TEST_DSN"], + user=os.environ["ORACLE_TEST_USER"], + password=os.environ["ORACLE_TEST_PASSWORD"], + ) as store: + collection = store.get_collection(dict, definition=definition, collection_name=table) + try: + assert not await collection.collection_exists() + await collection.ensure_collection_exists() + await collection.ensure_collection_exists() + assert table in await store.list_collection_names() + first = { + "id": "first", + "text": "100%_! literal", + "amount": 1, + "flag": True, + "embedding": [1.0, 0, 0], + "second": [1.0, 0, 0], + } + second = { + "id": "second", + "text": "unrelated", + "amount": 2, + "flag": False, + "embedding": [0, 1.0, 0], + "second": [0, 1.0, 0], + } + assert await collection.upsert([first, second], generate_vectors=False) == ["first", "second"] + assert await collection.get(["second", "first", "second"], include_vectors=True) == [ + second, + first, + second, + ] + assert [ + item["id"] + for item in await collection.get(filter=Filter("amount", "gte", 1), order_by={"amount": False}, top=1) + ] == ["second"] + assert [item["id"] for item in await collection.get(filter=Filter("text", "contains_text", "%_"))] == [ + "first" + ] + results = [ + item + async for item in await collection.search( + vector=[1, 0, 0], + filter=Filter("flag", "eq", True), + score_threshold=0.1, + ) + ] + assert [item["record"]["id"] for item in results] == ["first"] + assert results[0]["score"] == pytest.approx(0.0) + dot_results = [ + item + async for item in await collection.search( + vector=[1, 0, 0], + vector_property_name="second", + score_threshold=0.5, + ) + ] + assert [item["record"]["id"] for item in dot_results] == ["first"] + assert dot_results[0]["score"] == pytest.approx(1.0) + await collection.upsert([dict(first, text="updated")], generate_vectors=False) + assert (await collection.get(["first"]))[0]["text"] == "updated" + await collection.delete(["first", "second"]) + assert await collection.get() == [] + finally: + await collection.ensure_collection_deleted() + assert not await collection.collection_exists() diff --git a/python/packages/oracle/tests/oracle/test_settings.py b/python/packages/oracle/tests/oracle/test_settings.py new file mode 100644 index 00000000000..547890543c7 --- /dev/null +++ b/python/packages/oracle/tests/oracle/test_settings.py @@ -0,0 +1,183 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +from functools import partial +from typing import get_type_hints +from unittest.mock import AsyncMock, MagicMock, patch + +import oracledb +import pytest +from agent_framework import SecretString, load_settings +from agent_framework.exceptions import IntegrationException + +from agent_framework_oracle import OracleCollection, OracleSettings, OracleStore +from agent_framework_oracle import _vector_store as module + + +@pytest.fixture(params=["store", "collection"]) +def constructor(request, definition_factory, monkeypatch): + for name in ("ORACLE_DSN", "ORACLE_USER", "ORACLE_PASSWORD"): + monkeypatch.delenv(name, raising=False) + if request.param == "collection": + return partial(OracleCollection, dict, definition=definition_factory()) + return OracleStore + + +def test_public_settings_mask_password(monkeypatch): + monkeypatch.setenv("ORACLE_DSN", "localhost/service") + monkeypatch.setenv("ORACLE_USER", "example") + monkeypatch.setenv("ORACLE_PASSWORD", "private-value") + assert get_type_hints(OracleSettings) == { + "dsn": str | None, + "user": str | None, + "password": SecretString | None, + } + settings = load_settings(OracleSettings, env_prefix="ORACLE_") + assert settings["dsn"] == "localhost/service" + password = settings["password"] + assert isinstance(password, SecretString) + assert password.get_secret_value() == "private-value" + assert "private-value" not in repr(settings) + assert "private-value" not in str(password) + + +@pytest.mark.parametrize("wrapped", [False, True]) +def test_explicit_credentials_precede_file_and_environment(constructor, monkeypatch, tmp_path, wrapped): + monkeypatch.setenv("ORACLE_DSN", "environment") + monkeypatch.setenv("ORACLE_USER", "environment") + monkeypatch.setenv("ORACLE_PASSWORD", "environment") + env_file = tmp_path / "oracle.env" + env_file.write_text("ORACLE_DSN=file\nORACLE_USER=file\nORACLE_PASSWORD=file\n", encoding="utf-8") + secret = SecretString("explicit-password") if wrapped else "explicit-password" + instance = constructor( + dsn="explicit", + user="explicit", + password=secret, + env_file_path=str(env_file), + ) + assert instance._client._dsn == instance._client._user == "explicit" + assert instance._client._password.get_secret_value() == "explicit-password" + assert instance._client._pool is None + assert instance.managed_client + + +def test_selected_file_precedes_environment_and_supports_encoding(constructor, monkeypatch, tmp_path): + for name in ("ORACLE_DSN", "ORACLE_USER", "ORACLE_PASSWORD"): + monkeypatch.setenv(name, "environment") + env_file = tmp_path / "oracle.env" + env_file.write_text("ORACLE_DSN=file\nORACLE_USER=file\nORACLE_PASSWORD=file\n", encoding="utf-16") + instance = constructor(env_file_path=str(env_file), env_file_encoding="utf-16") + assert instance._client._dsn == instance._client._user == "file" + assert instance._client._password.get_secret_value() == "file" + + +def test_environment_without_implicit_dotenv(constructor, monkeypatch, tmp_path): + for name in ("ORACLE_DSN", "ORACLE_USER", "ORACLE_PASSWORD"): + monkeypatch.setenv(name, "environment") + (tmp_path / ".env").write_text("ORACLE_DSN=must-not-read\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + with patch("agent_framework._settings.dotenv_values", side_effect=AssertionError("Unexpected .env read")): + instance = constructor() + assert instance._client._dsn == "environment" + assert instance._client._pool is None + + +@pytest.mark.parametrize("missing", ["dsn", "user", "password"]) +def test_missing_or_empty_credentials_fail_without_connecting(constructor, missing): + settings = {"dsn": "dsn", "user": "user", "password": "password"} + settings[missing] = "" + with patch.object(module.oracledb, "create_pool_async") as create, pytest.raises(ValueError): + constructor(**settings) + create.assert_not_called() + + +def test_missing_selected_file_is_not_ignored(constructor, tmp_path): + with pytest.raises(FileNotFoundError): + constructor(dsn="explicit", user="explicit", password="explicit", env_file_path=str(tmp_path / "missing.env")) + + +@pytest.mark.parametrize("option", ["dsn", "user", "password", "env_file_path", "env_file_encoding"]) +def test_borrowed_client_rejects_connection_options(constructor, option): + client = MagicMock(spec=oracledb.AsyncConnection) + with pytest.raises(ValueError, match="cannot be combined"): + constructor(client=client, **{option: "not-read"}) + + +@pytest.mark.parametrize("client_type", [oracledb.AsyncConnection, oracledb.AsyncConnectionPool]) +async def test_borrowed_client_bypasses_settings_and_remains_open(constructor, client_type): + client = MagicMock(spec=client_type) + client.close = AsyncMock() + with patch.object(module, "load_settings", side_effect=AssertionError("Unexpected settings read")): + instance = constructor(client=client) + assert not instance.managed_client + await instance.close() + await instance.close() + client.close.assert_not_awaited() + + +def test_invalid_client_type_is_rejected(constructor): + with pytest.raises(TypeError, match="client"): + constructor(client=object()) + + +async def test_store_shares_client_without_rereading_settings(monkeypatch, definition_factory): + monkeypatch.setenv("ORACLE_DSN", "one") + monkeypatch.setenv("ORACLE_USER", "user") + monkeypatch.setenv("ORACLE_PASSWORD", "password") + store = OracleStore() + monkeypatch.setenv("ORACLE_DSN", "two") + with patch.object(module, "load_settings", side_effect=AssertionError("Unexpected settings read")): + collection = store.get_collection(dict, definition=definition_factory()) + assert collection._client is store._client + assert not collection.managed_client + await collection.close() + assert not store._client._closed + await store.close() + assert store._client._closed + + +async def test_owned_pool_is_lazy_and_closed_once(): + pool = MagicMock(spec=oracledb.AsyncConnectionPool) + pool.close = AsyncMock() + store = OracleStore(dsn="dsn", user="user", password=SecretString("password")) + with patch.object(module.oracledb, "create_pool_async", return_value=pool) as create: + async with store: + assert store._client._pool is None + assert await store._client._get_client() is pool + assert await store._client._get_client() is pool + await store.close() + create.assert_called_once() + pool.close.assert_awaited_once() + with pytest.raises(RuntimeError, match="closed"): + await store._client._get_client() + + +async def test_pooled_writes_commit_or_rollback_but_borrowed_connection_does_not(): + connection = MagicMock(spec=oracledb.AsyncConnection) + connection.commit = AsyncMock() + connection.rollback = AsyncMock() + pool = MagicMock(spec=oracledb.AsyncConnectionPool) + pool.acquire.return_value.__aenter__.return_value = connection + for client in (pool, connection): + owner = module._Client(client=client) + async with owner.connection(write=True): + pass + try: + async with owner.connection(write=True): + raise ValueError("caller failure") + except ValueError: + pass + connection.commit.assert_awaited_once() + connection.rollback.assert_awaited_once() + assert pool.acquire.call_count == 2 + + +async def test_driver_connection_failure_is_chained_without_leaking_credentials(): + pool = MagicMock(spec=oracledb.AsyncConnectionPool) + pool.acquire.side_effect = oracledb.OperationalError("connection refused") + owner = module._Client(client=pool) + with pytest.raises(IntegrationException, match="Oracle operation failed") as error: + async with owner.connection(): + pytest.fail("Acquisition should fail before yielding.") + assert isinstance(error.value.__cause__, oracledb.OperationalError) diff --git a/python/pyproject.toml b/python/pyproject.toml index 35066954f0e..3526050a6c4 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -104,6 +104,7 @@ agent-framework-mongodb = { workspace = true } agent-framework-monty = { workspace = true } agent-framework-ollama = { workspace = true } agent-framework-openai = { workspace = true } +agent-framework-oracle = { workspace = true } agent-framework-orchestrations = { workspace = true } agent-framework-postgres = { workspace = true } agent-framework-purview = { workspace = true } diff --git a/python/uv.lock b/python/uv.lock index 67edc89074c..1a2945870c4 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -62,6 +62,7 @@ members = [ "agent-framework-monty", "agent-framework-ollama", "agent-framework-openai", + "agent-framework-oracle", "agent-framework-orchestrations", "agent-framework-postgres", "agent-framework-purview", @@ -917,6 +918,21 @@ requires-dist = [ { name = "openai", specifier = ">=2.25.0,<4" }, ] +[[package]] +name = "agent-framework-oracle" +version = "1.0.0a260923" +source = { editable = "packages/oracle" } +dependencies = [ + { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "oracledb", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] + +[package.metadata] +requires-dist = [ + { name = "agent-framework-core", editable = "packages/core" }, + { name = "oracledb", specifier = ">=2.2.0,<4" }, +] + [[package]] name = "agent-framework-orchestrations" version = "1.2.0" @@ -4385,6 +4401,38 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/23/3f/ab8d29df207ce5f470a07fa96ebb48af4e95b7fab7e7635311b9a32f2fab/opentelemetry_util_http-0.65b0-py3-none-any.whl", hash = "sha256:7553b606f963097cb190536dc30556cce85090692e471a422fff30ca29b04348", size = 8245, upload-time = "2026-07-16T15:25:46.482Z" }, ] +[[package]] +name = "oracledb" +version = "3.4.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "typing-extensions", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f7/02/70a872d1a4a739b4f7371ab8d3d5ed8c6e57e142e2503531aafcb220893c/oracledb-3.4.2.tar.gz", hash = "sha256:46e0f2278ff1fe83fbc33a3b93c72d429323ec7eed47bc9484e217776cd437e5", size = 855467, upload-time = "2026-01-28T17:25:39.91Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/64/80/be263b668ba32b258d07c85f7bfb6967a9677e016c299207b28734f04c4b/oracledb-3.4.2-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:b8e4b8a852251cef09038b75f30fce1227010835f4e19cfbd436027acba2697c", size = 4228552, upload-time = "2026-01-28T17:25:54.844Z" }, + { url = "https://files.pythonhosted.org/packages/91/bc/e832a649529da7c60409a81be41f3213b4c7ffda4fe424222b2145e8d43c/oracledb-3.4.2-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1617a1db020346883455af005efbefd51be2c4d797e43b1b38455a19f8526b48", size = 2421924, upload-time = "2026-01-28T17:25:56.984Z" }, + { url = "https://files.pythonhosted.org/packages/86/21/d867c37e493a63b5521bd248110ad5b97b18253d64a30703e3e8f3d9631e/oracledb-3.4.2-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5ed78d7e7079a778062744ccf42141ce4806818c3f4dd6463e4a7edd561c9f86", size = 2599301, upload-time = "2026-01-28T17:25:58.529Z" }, + { url = "https://files.pythonhosted.org/packages/2a/de/9b1843ea27f7791449652d7f340f042c3053336d2c11caf29e59bab86189/oracledb-3.4.2-cp311-cp311-win32.whl", hash = "sha256:0e16fe3d057e0c41a23ad2ae95bfa002401690773376d476be608f79ac74bf05", size = 1492890, upload-time = "2026-01-28T17:26:00.662Z" }, + { url = "https://files.pythonhosted.org/packages/d6/10/cbc8afa2db0cec80530858d3e4574f9734fae8c0b7f1df261398aa026c5f/oracledb-3.4.2-cp311-cp311-win_amd64.whl", hash = "sha256:f93cae08e8ed20f2d5b777a8602a71f9418389c661d2c937e84d94863e7e7011", size = 1843355, upload-time = "2026-01-28T17:26:02.637Z" }, + { url = "https://files.pythonhosted.org/packages/8f/81/2e6154f34b71cd93b4946c73ea13b69d54b8d45a5f6bbffe271793240d21/oracledb-3.4.2-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:a7396664e592881225ba66385ee83ce339d864f39003d6e4ca31a894a7e7c552", size = 4220806, upload-time = "2026-01-28T17:26:04.322Z" }, + { url = "https://files.pythonhosted.org/packages/ab/a9/a1d59aaac77d8f727156ec6a3b03399917c90b7da4f02d057f92e5601f56/oracledb-3.4.2-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0f04a2d62073407672f114d02529921de0677c6883ed7c64d8d1a3c04caa3238", size = 2233795, upload-time = "2026-01-28T17:26:05.877Z" }, + { url = "https://files.pythonhosted.org/packages/94/ec/8c4a38020cd251572bd406ddcbde98ca052ec94b5684f9aa9ef1ddfcc68c/oracledb-3.4.2-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d8d75e4f879b908be66cce05ba6c05791a5dbb4a15e39abc01aa25c8a2492bd9", size = 2424756, upload-time = "2026-01-28T17:26:07.35Z" }, + { url = "https://files.pythonhosted.org/packages/fa/7d/c251c2a8567151ccfcfbe3467ea9a60fb5480dc4719342e2e6b7a9679e5d/oracledb-3.4.2-cp312-cp312-win32.whl", hash = "sha256:31b7ee83c23d0439778303de8a675717f805f7e8edb5556d48c4d8343bcf14f5", size = 1453486, upload-time = "2026-01-28T17:26:08.869Z" }, + { url = "https://files.pythonhosted.org/packages/4c/78/c939f3c16fb39400c4734d5a3340db5659ba4e9dce23032d7b33ccfd3fe5/oracledb-3.4.2-cp312-cp312-win_amd64.whl", hash = "sha256:ac25a0448fc830fb7029ad50cd136cdbfcd06975d53967e269772cc5cb8c203a", size = 1794445, upload-time = "2026-01-28T17:26:10.66Z" }, + { url = "https://files.pythonhosted.org/packages/22/68/f7126f5d911c295b57720c6b1a0609a5a2667b4546946433552a4de46333/oracledb-3.4.2-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:643c25d301a289a371e37fcedb59e5fa5e54fb321708e5c12821c4b55bdd8a4d", size = 4205176, upload-time = "2026-01-28T17:26:12.463Z" }, + { url = "https://files.pythonhosted.org/packages/5d/93/2fced60f92dc82e66980a8a3ba5c1ea48110bf1dd81d030edb69d88f992e/oracledb-3.4.2-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:55397e7eb43bb7017c03a981c736c25724182f5210951181dfe3fab0e5d457fb", size = 2231298, upload-time = "2026-01-28T17:26:14.497Z" }, + { url = "https://files.pythonhosted.org/packages/75/a7/4dd286f3a6348d786fef9e6ab2e6c9b74ca9195d9a756f2a67e45743cdf0/oracledb-3.4.2-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b26a10f9c790bd141ffc8af68520803ed4a44a9258bf7d1eea9bfdd36bd6df7f", size = 2439430, upload-time = "2026-01-28T17:26:16.044Z" }, + { url = "https://files.pythonhosted.org/packages/19/28/94bc753e5e969c60ee5d9c914e2b4ef79999eaca8e91bcab2fbf0586b80b/oracledb-3.4.2-cp313-cp313-win32.whl", hash = "sha256:b974caec2c330c22bbe765705a5ac7d98ec3022811dec2042d561a3c65cb991b", size = 1458209, upload-time = "2026-01-28T17:26:17.652Z" }, + { url = "https://files.pythonhosted.org/packages/cb/2b/593a9b2d4c12c9de3289e67d84fe023336d99f36ba51442a5a0f5ce6acf7/oracledb-3.4.2-cp313-cp313-win_amd64.whl", hash = "sha256:3df8eee1410d25360599968b1625b000f10c5ae0e47274031a7842a9dc418890", size = 1793558, upload-time = "2026-01-28T17:26:19.914Z" }, + { url = "https://files.pythonhosted.org/packages/42/20/1e98f84c1555911c46b4fa870fbef2a80617bf7e0a5f178078ecf466c917/oracledb-3.4.2-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:59ad6438f56a25e8e1a4a3dd1b42235a5d09ab9ba417ff2ad14eae6596f3d06f", size = 4247459, upload-time = "2026-01-28T17:26:22.356Z" }, + { url = "https://files.pythonhosted.org/packages/7d/74/95963e2d94f84b9937a562a9a2529f72d050afbc2ffd88f6661e3a876f7d/oracledb-3.4.2-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:404ec1451d0448653ee074213b87d6c5bd65eaa74b50083ddf2c9c3e11c71c71", size = 2271749, upload-time = "2026-01-28T17:26:24.078Z" }, + { url = "https://files.pythonhosted.org/packages/82/89/38ce85148a246087795379ee52c5b20726a00a69c87ba6ec266bcdad30fc/oracledb-3.4.2-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:19fa80ef84f85ad74077aa626067bbe697e527bd39604b4209f9d86cb2876b89", size = 2452031, upload-time = "2026-01-28T17:26:26.08Z" }, + { url = "https://files.pythonhosted.org/packages/3f/8d/51fe907fdec0267ad7c6e9a62998cbe878efcd168ea6e39f162fab62fdaa/oracledb-3.4.2-cp314-cp314-win32.whl", hash = "sha256:d7ce75c498bff758548ec6e4424ab4271aa257e5887cc436a54bc947fd46199a", size = 1480973, upload-time = "2026-01-28T17:26:27.584Z" }, + { url = "https://files.pythonhosted.org/packages/48/22/a37354f19786774e5e4041338043b516db060aacfdfcd5aca8bb92c2539a/oracledb-3.4.2-cp314-cp314-win_amd64.whl", hash = "sha256:5d7befb014174c5ae11c3a08f5ed6668a25ab2335d8e7104dca70d54d54a5b3a", size = 1837756, upload-time = "2026-01-28T17:26:29.032Z" }, +] + [[package]] name = "orderedmultidict" version = "1.0.2" From 7d16683dc89b913454b3f5c15a05ac0dac7f5aca Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 23 Sep 2026 13:43:32 +0200 Subject: [PATCH 2/3] fix(python): preserve Oracle filter semantics and test typing --- python/packages/oracle/README.md | 7 ++- .../agent_framework_oracle/_vector_store.py | 36 +++++++---- .../packages/oracle/tests/oracle/conftest.py | 2 +- .../oracle/tests/oracle/test_collection.py | 22 +++++++ .../oracle/tests/oracle/test_filters.py | 63 +++++++++++++++++++ .../oracle/tests/oracle/test_settings.py | 10 +++ 6 files changed, 124 insertions(+), 16 deletions(-) diff --git a/python/packages/oracle/README.md b/python/packages/oracle/README.md index a93b340a2fb..a5e799e6888 100644 --- a/python/packages/oracle/README.md +++ b/python/packages/oracle/README.md @@ -118,8 +118,11 @@ creates and deletes a unique test table. - Portable filters support scalar equality (including `NULL` and boolean distinctions), `in`/`not_in`, numeric ranges, `is_null`, `is_not_null`, `exists`, literal text `contains_text`/`starts_with`/`ends_with`, and - AND/OR/NOT. Collection membership and nested filters are rejected. - Ordered filters on integer fields reject non-integral float operands. + AND/OR/NOT. Finite `Decimal` values are supported for exact numeric equality + and membership; malformed UUID filter values compare as unequal. + Collection membership and nested filters are rejected. + Ordered filters on integer fields reject non-integral float operands; + ordered `Decimal` operands are not supported. String comparison follows the configured Oracle collation. Input values are bound; identifiers are checked and quoted as individual names. diff --git a/python/packages/oracle/agent_framework_oracle/_vector_store.py b/python/packages/oracle/agent_framework_oracle/_vector_store.py index 2f73f3199e6..b0579588508 100644 --- a/python/packages/oracle/agent_framework_oracle/_vector_store.py +++ b/python/packages/oracle/agent_framework_oracle/_vector_store.py @@ -345,30 +345,40 @@ def _equality(self, field: VectorStoreField, value: Any) -> str: if value is None: return f"{column} IS NULL" kind = field.type_ - if kind in ("int", "float") and type(value) in (int, float): + if kind in ("int", "float") and (type(value) in (int, float) or isinstance(value, Decimal)): if type(value) is float and not math.isfinite(value): raise ValueError("Numeric filter values must be finite.") + if isinstance(value, Decimal) and not value.is_finite(): + raise ValueError("Numeric filter values must be finite.") if kind == "int": - if type(value) is float: - if not value.is_integer() or not -(2**63) <= value < 2**63: - return "1=0" - value = int(value) - elif not -(2**63) <= value < 2**63: + if not -(2**63) <= value < 2**63 or value != int(value): return "1=0" - elif type(value) is int: + adapted = int(value) + else: try: - numerator, denominator = float(value).as_integer_ratio() - if numerator != value * denominator: - return "1=0" + adapted = float(value) except OverflowError: return "1=0" - adapted = value if kind == "int" else float(value) + if not math.isfinite(adapted): + return "1=0" + if isinstance(value, Decimal) and Decimal.from_float(adapted) != value: + return "1=0" + if type(value) is int: + numerator, denominator = adapted.as_integer_ratio() + if numerator != value * denominator: + return "1=0" + elif kind == "UUID": + if not isinstance(value, UUID | str): + return "1=0" + try: + adapted = _validate_scalar(field, value) + except ValueError: + return "1=0" elif ( (kind == "bool" and type(value) is not bool) or (kind != "bool" and isinstance(value, bool)) - or (kind == "UUID" and not isinstance(value, UUID | str)) or (kind == "str" and not isinstance(value, str)) - or (kind in ("int", "float") and type(value) not in (int, float)) + or (kind in ("int", "float") and type(value) not in (int, float, Decimal)) ): return "1=0" else: diff --git a/python/packages/oracle/tests/oracle/conftest.py b/python/packages/oracle/tests/oracle/conftest.py index ac94102ca12..eea10e7cde4 100644 --- a/python/packages/oracle/tests/oracle/conftest.py +++ b/python/packages/oracle/tests/oracle/conftest.py @@ -101,7 +101,7 @@ def mock_database(collection): cursor = FakeCursor() connection = MagicMock() connection.cursor.return_value = cursor - writes = [] + writes: list[bool] = [] @asynccontextmanager async def acquire(*, write=False): diff --git a/python/packages/oracle/tests/oracle/test_collection.py b/python/packages/oracle/tests/oracle/test_collection.py index 30461549fda..228a2bb675b 100644 --- a/python/packages/oracle/tests/oracle/test_collection.py +++ b/python/packages/oracle/tests/oracle/test_collection.py @@ -265,6 +265,15 @@ async def test_retrieval_projects_vectors_only_when_requested(collection, mock_d assert '"dense vector"' in cursor.execute.call_args.args[0] +async def test_get_applies_decimal_equality_after_core_filter_validation(collection, mock_database): + _, cursor, _, _ = mock_database + cursor.fetchall.return_value = [("one", "text", 2, 1)] + assert (await collection.get(filter=Filter("number", "eq", Decimal("2"))))[0]["number"] == 2 + statement, binds = cursor.execute.call_args.args + assert '"number" = :f0' in statement + assert binds["f0"] == 2 + + async def test_key_lookup_preserves_order_duplicates_and_chunks(collection, mock_database): _, cursor, _, _ = mock_database cursor.fetchall.side_effect = [ @@ -329,6 +338,19 @@ async def test_search_filter_threshold_and_paging_execute_in_database(collection cursor.setinputsizes.assert_called_once_with(query_vector=oracledb.DB_TYPE_VECTOR) +async def test_search_applies_invalid_uuid_as_non_equal_after_core_filter_validation(definition_factory, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, definition=definition_factory(key_type="UUID"), dsn="unused", user="user", password="pass" + ) + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + results = await collection.search(vector=[1, 0, 0], filter=Filter("id", "in", ["not-a-uuid"])) + assert [item async for item in results] == [] + statement, binds = cursor.execute.call_args.args + assert '"doc""id" IS NOT NULL AND (1=0)' in statement + assert set(binds) == {"query_vector", "top", "skip"} + + @pytest.mark.parametrize( "metric,cutoff,score_expression", [ diff --git a/python/packages/oracle/tests/oracle/test_filters.py b/python/packages/oracle/tests/oracle/test_filters.py index bdfe96da74a..0f4967e5ab3 100644 --- a/python/packages/oracle/tests/oracle/test_filters.py +++ b/python/packages/oracle/tests/oracle/test_filters.py @@ -2,6 +2,9 @@ from __future__ import annotations +from decimal import Decimal +from uuid import UUID + import pytest from agent_framework import Filter, FilterGroup @@ -55,6 +58,66 @@ def test_numeric_equality_does_not_coerce_invalid_values(collection, value): assert sql == "(NOT (1=0))" and binds == {} +@pytest.mark.parametrize( + ("field", "value", "expected_bind"), + [ + ("number", Decimal("2"), 2), + ("number", Decimal("2.0"), 2), + ("number", Decimal("9223372036854775807"), 2**63 - 1), + ("number", Decimal("2.5"), None), + ("number", Decimal("9223372036854775808"), None), + ("text", Decimal("0.5"), 0.5), + ("text", Decimal("0.1"), None), + ("text", Decimal("1e1000"), None), + ], +) +def test_decimal_equality_uses_exact_numeric_comparison(definition_factory, field, value, expected_bind): + compiler = _FilterCompiler(definition_factory(data_type="float")) + for operator in ("eq", "ne"): + sql, binds = compiler.compile(Filter(field, operator, value)) + if expected_bind is None: + assert sql == ("1=0" if operator == "eq" else "(NOT (1=0))") + assert binds == {} + else: + assert f'"{field if field == "number" else "body text"}" = :f0' in sql + assert binds == {"f0": expected_bind} + + +def test_decimal_membership_and_nonfinite_values(collection): + sql, binds = collection._prepare_filter(Filter("number", "in", [Decimal("2"), Decimal("2.5")])) + assert '"number" = :f0' in sql + assert binds == {"f0": 2} + sql, binds = collection._prepare_filter(Filter("number", "not_in", [Decimal("2.5")])) + assert sql == '("number" IS NOT NULL AND (NOT ((1=0))))' + assert binds == {} + with pytest.raises(ValueError, match="finite"): + collection._prepare_filter(Filter("number", "eq", Decimal("NaN"))) + + +@pytest.mark.parametrize( + ("operator", "value", "expected_sql"), + [ + ("eq", "not-a-uuid", "1=0"), + ("ne", "not-a-uuid", "(NOT (1=0))"), + ("in", ["not-a-uuid"], '("doc""id" IS NOT NULL AND (1=0))'), + ("not_in", ["not-a-uuid"], '("doc""id" IS NOT NULL AND (NOT ((1=0))))'), + ], +) +def test_malformed_uuid_filter_is_non_equal(definition_factory, operator, value, expected_sql): + compiler = _FilterCompiler(definition_factory(key_type="UUID")) + sql, binds = compiler.compile(Filter("id", operator, value)) + assert sql == expected_sql + assert binds == {} + + +def test_valid_uuid_filter_values_remain_bound(definition_factory): + compiler = _FilterCompiler(definition_factory(key_type="UUID")) + valid = UUID(int=2) + sql, binds = compiler.compile(Filter("id", "in", ["malformed", valid])) + assert '"doc""id" = :f0' in sql + assert binds == {"f0": str(valid)} + + def test_boolean_equality_does_not_coerce_numbers(collection): sql, binds = collection._prepare_filter(Filter("flag", "eq", True)) assert '"flag" = :f0' in sql and binds == {"f0": 1} diff --git a/python/packages/oracle/tests/oracle/test_settings.py b/python/packages/oracle/tests/oracle/test_settings.py index 547890543c7..d34bd30cc7b 100644 --- a/python/packages/oracle/tests/oracle/test_settings.py +++ b/python/packages/oracle/tests/oracle/test_settings.py @@ -153,6 +153,16 @@ async def test_owned_pool_is_lazy_and_closed_once(): await store._client._get_client() +async def test_owned_pool_uses_driver_factory_without_connecting(): + store = OracleStore(dsn="localhost:1521/unused", user="unused", password="unused") + try: + pool = await store._client._get_client() + assert isinstance(pool, oracledb.AsyncConnectionPool) + assert await store._client._get_client() is pool + finally: + await store.close() + + async def test_pooled_writes_commit_or_rollback_but_borrowed_connection_does_not(): connection = MagicMock(spec=oracledb.AsyncConnection) connection.commit = AsyncMock() From 5e8edfc08c4669a694900e06dd152cdc37dd0292 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 30 Sep 2026 09:35:46 +0200 Subject: [PATCH 3/3] Python: Fix Oracle native vector retrieval --- .../agent_framework_oracle/_vector_store.py | 3 + .../oracle/tests/oracle/test_collection.py | 63 +++++++++++++++++++ .../oracle/tests/oracle/test_integration.py | 4 +- 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/python/packages/oracle/agent_framework_oracle/_vector_store.py b/python/packages/oracle/agent_framework_oracle/_vector_store.py index b0579588508..465f8356ab5 100644 --- a/python/packages/oracle/agent_framework_oracle/_vector_store.py +++ b/python/packages/oracle/agent_framework_oracle/_vector_store.py @@ -616,6 +616,9 @@ def _decode_row(self, fields: Sequence[VectorStoreField], row: Sequence[Any]) -> raise IntegrationInvalidResponseException("Oracle returned an unexpected number of columns.") result: dict[str, Any] = {} for field, value in zip(fields, row, strict=True): + if field.field_type == "vector": + result[field.storage_name or field.name] = value + continue if value is not None and field.type_ == "UUID": if not isinstance(value, UUID | str): raise IntegrationInvalidResponseException(f"Oracle returned an invalid UUID for '{field.name}'.") diff --git a/python/packages/oracle/tests/oracle/test_collection.py b/python/packages/oracle/tests/oracle/test_collection.py index 228a2bb675b..dde235c53de 100644 --- a/python/packages/oracle/tests/oracle/test_collection.py +++ b/python/packages/oracle/tests/oracle/test_collection.py @@ -427,6 +427,69 @@ def test_typed_model_deserializes_native_vectors(): assert record.embedding == [1, 0, 0] +@pytest.mark.parametrize("by_key", [True, False]) +@pytest.mark.parametrize("vector", [array("f", [1, 0, 0]), None]) +async def test_typed_model_get_deserializes_native_vectors(by_key, vector, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection(TypedDocument, dsn="unused", user="user", password="example") + assert collection.definition.vector_fields[0].type_ == "float" + key = UUID(int=1) + cursor.fetchall.return_value = [(str(key), "one", vector)] + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + if by_key: + records = await collection.get([key], include_vectors=True) + else: + records = await collection.get(filter=Filter("text", "eq", "one"), include_vectors=True) + assert records == [TypedDocument(key, "one", [1.0, 0.0, 0.0] if vector is not None else None)] + + +@pytest.mark.parametrize("include_vectors", [True, False]) +async def test_typed_model_search_deserializes_native_vectors(include_vectors, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection(TypedDocument, dsn="unused", user="user", password="example") + key = UUID(int=1) + row = (str(key), "one", array("f", [1, 0, 0]), 0.0) if include_vectors else (str(key), "one", 0.0) + cursor.fetchall.return_value = [row] + with patch.object(collection._client, "connection", side_effect=acquire.side_effect): + results = await collection.search(vector=[1, 0, 0], include_vectors=include_vectors) + assert [item async for item in results] == [ + { + "record": TypedDocument(key, "one", [1.0, 0.0, 0.0] if include_vectors else None), + "score": 0.0, + } + ] + + +@pytest.mark.parametrize("vector", [[1.0, 0.0, 0.0], 1.0]) +async def test_typed_model_get_rejects_invalid_native_vectors(vector, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection(TypedDocument, dsn="unused", user="user", password="example") + key = UUID(int=1) + cursor.fetchall.return_value = [(str(key), "one", vector)] + with ( + patch.object(collection._client, "connection", side_effect=acquire.side_effect), + pytest.raises(IntegrationInvalidResponseException, match="invalid vector"), + ): + await collection.get([key], include_vectors=True) + + +async def test_get_rejects_native_vector_in_scalar_float_column(definition_factory, mock_database): + _, cursor, _, acquire = mock_database + collection = OracleCollection( + dict, + definition=definition_factory(data_type="float"), + dsn="unused", + user="user", + password="example", + ) + cursor.fetchall.return_value = [("one", array("f", [1, 0, 0]), 2, 1)] + with ( + patch.object(collection._client, "connection", side_effect=acquire.side_effect), + pytest.raises(IntegrationInvalidResponseException, match="invalid float"), + ): + await collection.get(["one"]) + + async def test_uuid_key_lookup_uses_normalized_bind(definition_factory, mock_database): _, cursor, _, acquire = mock_database collection = OracleCollection( diff --git a/python/packages/oracle/tests/oracle/test_integration.py b/python/packages/oracle/tests/oracle/test_integration.py index 0ec4ee86c75..18519721b89 100644 --- a/python/packages/oracle/tests/oracle/test_integration.py +++ b/python/packages/oracle/tests/oracle/test_integration.py @@ -31,7 +31,7 @@ async def test_oracle_vector_lifecycle_crud_filters_and_search(): VectorStoreField("data", name="text", type_="str", storage_name="BODY"), VectorStoreField("data", name="amount", type_="int"), VectorStoreField("data", name="flag", type_="bool"), - VectorStoreField("vector", name="embedding", type_="float32", dimensions=3), + VectorStoreField("vector", name="embedding", type_="float", dimensions=3), VectorStoreField("vector", name="second", type_="float64", dimensions=3, distance_function="dot_prod"), ]) table = f"AF_ORACLE_TEST_{uuid4().hex}" @@ -91,9 +91,11 @@ async def test_oracle_vector_lifecycle_crud_filters_and_search(): vector=[1, 0, 0], vector_property_name="second", score_threshold=0.5, + include_vectors=True, ) ] assert [item["record"]["id"] for item in dot_results] == ["first"] + assert dot_results[0]["record"] == first assert dot_results[0]["score"] == pytest.approx(1.0) await collection.upsert([dict(first, text="updated")], generate_vectors=False) assert (await collection.get(["first"]))[0]["text"] == "updated"