From d9efe4e60e91c1fecf1dcd6a69d4f752a277fb2e Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 17:43:17 +0000 Subject: [PATCH 01/15] test(adbc): cover opaque UUID row decoding --- .../adapters/test_adbc/test_uuid_decoding.py | 122 ++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 tests/unit/adapters/test_adbc/test_uuid_decoding.py diff --git a/tests/unit/adapters/test_adbc/test_uuid_decoding.py b/tests/unit/adapters/test_adbc/test_uuid_decoding.py new file mode 100644 index 000000000..6e11b58d8 --- /dev/null +++ b/tests/unit/adapters/test_adbc/test_uuid_decoding.py @@ -0,0 +1,122 @@ +"""Unit tests for PostgreSQL-family ADBC UUID result decoding.""" + +from typing import TYPE_CHECKING, Any, cast +from uuid import UUID + +import pyarrow as pa + +from sqlspec.adapters.adbc.core import get_statement_config +from sqlspec.adapters.adbc.driver import AdbcDriver + +if TYPE_CHECKING: + from sqlspec.adapters.adbc._typing import AdbcConnection + + +UUID_VALUE = UUID("550e8400-e29b-41d4-a716-446655440000") +OTHER_UUID_VALUE = UUID("550e8400-e29b-41d4-a716-446655440001") + + +def test_opaque_uuid_columns_decode_scalar_list_and_null_values() -> None: + table = _build_uuid_table() + driver = _make_driver(table) + + rows = driver.select("SELECT identifier, identifiers, label FROM uuid_values") + + assert rows == [ + {"identifier": UUID_VALUE, "identifiers": [UUID_VALUE, None, OTHER_UUID_VALUE], "label": "first"}, + {"identifier": None, "identifiers": [], "label": "second"}, + ] + + +def test_disabled_arrow_extension_types_preserve_opaque_storage_bytes() -> None: + table = _build_uuid_table() + driver = _make_driver(table, enable_arrow_extension_types=False) + + rows = driver.select("SELECT identifier, identifiers, label FROM uuid_values") + + assert rows == table.to_pylist() + + +def test_plain_arrow_table_uses_whole_table_pylist_path() -> None: + table = _TrackingTable(pa.table({"identifier": [1, 2], "label": ["first", "second"]})) + driver = _make_driver(cast("pa.Table", table)) + + rows = driver.select("SELECT identifier, label FROM plain_values") + + assert rows == [{"identifier": 1, "label": "first"}, {"identifier": 2, "label": "second"}] + assert table.to_pylist_calls == 1 + + +class _AdbcUuidResultCursor: + def __init__(self, table: pa.Table) -> None: + self.closed = False + self.executed: list[tuple[str, object]] = [] + self._table = table + + def execute(self, sql: str, parameters: object = None) -> None: + self.executed.append((sql, parameters)) + + def fetch_arrow_table(self) -> pa.Table: + return self._table + + def close(self) -> None: + self.closed = True + + +class _AdbcUuidResultConnection: + def __init__(self, table: pa.Table) -> None: + self.cursor_obj = _AdbcUuidResultCursor(table) + + def adbc_get_info(self) -> dict[str, str]: + return {"vendor_name": "postgres", "driver_name": "postgres"} + + def cursor(self) -> _AdbcUuidResultCursor: + return self.cursor_obj + + +class _TrackingTable: + def __init__(self, table: pa.Table) -> None: + self._table = table + self.to_pylist_calls = 0 + + @property + def column_names(self) -> list[str]: + return self._table.column_names + + @property + def schema(self) -> pa.Schema: + return self._table.schema + + def column(self, name: str) -> pa.ChunkedArray: + return self._table.column(name) + + def to_pylist(self) -> list[dict[str, Any]]: + self.to_pylist_calls += 1 + return self._table.to_pylist() + + +def _make_driver(table: pa.Table, *, enable_arrow_extension_types: bool = True) -> AdbcDriver: + connection = _AdbcUuidResultConnection(table) + return AdbcDriver( + cast("AdbcConnection", connection), + statement_config=get_statement_config("postgres"), + driver_features={"enable_arrow_extension_types": enable_arrow_extension_types}, + dialect="postgres", + ) + + +def _build_uuid_table() -> pa.Table: + opaque_type = pa.opaque(pa.binary(), "uuid", "PostgreSQL") + scalar_values = pa.ExtensionArray.from_storage( + opaque_type, + pa.array([UUID_VALUE.bytes, None], type=pa.binary()), + ) + list_values = pa.ExtensionArray.from_storage( + opaque_type, + pa.array([UUID_VALUE.bytes, None, OTHER_UUID_VALUE.bytes], type=pa.binary()), + ) + uuid_lists = pa.ListArray.from_arrays(pa.array([0, 3, 3], type=pa.int32()), list_values) + return pa.Table.from_arrays( + [scalar_values, uuid_lists, pa.array(["first", "second"])], + names=["identifier", "identifiers", "label"], + ) From 1870371c186b6aea346676a582a07a0b20e52bfa Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 17:47:46 +0000 Subject: [PATCH 02/15] fix(adbc): decode opaque UUID row results --- sqlspec/adapters/adbc/driver.py | 14 +++-- sqlspec/utils/arrow_helpers.py | 53 +++++++++++++++++-- .../adapters/test_adbc/test_uuid_decoding.py | 29 +++++++--- 3 files changed, 82 insertions(+), 14 deletions(-) diff --git a/sqlspec/adapters/adbc/driver.py b/sqlspec/adapters/adbc/driver.py index 374c7bee3..f910cfce4 100644 --- a/sqlspec/adapters/adbc/driver.py +++ b/sqlspec/adapters/adbc/driver.py @@ -40,7 +40,7 @@ ) from sqlspec.driver import BaseSyncExceptionHandler, SyncDriverAdapterBase, SyncRowStream from sqlspec.exceptions import DatabaseConnectionError, SQLSpecError -from sqlspec.utils.arrow_helpers import arrow_reader_with_deferred_close +from sqlspec.utils.arrow_helpers import arrow_reader_with_deferred_close, arrow_table_to_pylist from sqlspec.utils.logging import get_logger from sqlspec.utils.module_loader import ensure_pyarrow from sqlspec.utils.serializers import to_json @@ -129,7 +129,12 @@ def fetch_chunk(self) -> "list[dict[str, Any]]": batch = next(reader) except StopIteration: return [] - rows = cast("list[dict[str, Any]]", batch.to_pylist()) + rows = arrow_table_to_pylist( + batch, + decode_arrow_extension_types=bool( + self._driver.driver_features.get("enable_arrow_extension_types", True) + ), + ) if rows: return rows @@ -224,7 +229,10 @@ def dispatch_execute(self, cursor: "AdbcRawCursor", statement: SQL) -> "Executio if is_select_like: arrow_table = cursor.fetch_arrow_table() - data = arrow_table.to_pylist() + data = arrow_table_to_pylist( + arrow_table, + decode_arrow_extension_types=bool(self.driver_features.get("enable_arrow_extension_types", True)), + ) column_names = list(arrow_table.column_names) return self.create_execution_result( cursor, diff --git a/sqlspec/utils/arrow_helpers.py b/sqlspec/utils/arrow_helpers.py index 9c17c8d19..62593769d 100644 --- a/sqlspec/utils/arrow_helpers.py +++ b/sqlspec/utils/arrow_helpers.py @@ -11,6 +11,7 @@ from collections.abc import Callable, Iterable, Mapping from functools import lru_cache from typing import TYPE_CHECKING, Any, Literal, cast, overload +from uuid import UUID from sqlspec.exceptions import ImproperConfigurationError from sqlspec.utils.dispatch import TypeDispatcher @@ -385,9 +386,17 @@ def arrow_table_needs_parameter_preparation(table: "ArrowTable") -> bool: return _arrow_schema_needs_preparation(table.schema) -def arrow_table_to_pylist(table: "ArrowTable") -> "list[dict[str, Any]]": - """Convert Arrow table to list of dictionaries.""" - return table.to_pylist() +def arrow_table_to_pylist( + table: "ArrowTable | ArrowRecordBatch", *, decode_arrow_extension_types: bool = False +) -> "list[dict[str, Any]]": + """Convert Arrow data to dictionaries and optionally decode opaque UUID fields.""" + if not decode_arrow_extension_types or not _arrow_schema_has_opaque_uuid(table.schema): + return table.to_pylist() + + column_values = [ + _arrow_uuid_column_to_pylist(table.column(index), field.type) for index, field in enumerate(table.schema) + ] + return [dict(zip(table.column_names, row, strict=False)) for row in zip(*column_values, strict=False)] def arrow_table_column_names(table: "ArrowTable") -> "list[str]": @@ -507,3 +516,41 @@ def arrow_reader_to_return_format( table = reader.read_all() shaped = arrow_table_to_return_format(table, return_format=return_format, batch_size=batch_size) return shaped, int(table.num_rows) + + +def _arrow_schema_has_opaque_uuid(schema: Any) -> bool: + return any( + _arrow_type_is_opaque_uuid(field.type) or _arrow_type_is_list_of_opaque_uuid(field.type) for field in schema + ) + + +def _arrow_type_is_opaque_uuid(data_type: Any) -> bool: + return ( + getattr(data_type, "extension_name", None) == "arrow.opaque" + and getattr(data_type, "type_name", "").casefold() == "uuid" + ) + + +def _arrow_type_is_list_of_opaque_uuid(data_type: Any) -> bool: + ensure_pyarrow() + import pyarrow as pa + + is_list = pa.types.is_list(data_type) or pa.types.is_large_list(data_type) or pa.types.is_fixed_size_list(data_type) + return is_list and _arrow_type_is_opaque_uuid(data_type.value_type) + + +def _arrow_uuid_column_to_pylist(column: Any, data_type: Any) -> "list[Any]": + if _arrow_type_is_opaque_uuid(data_type): + ensure_pyarrow() + import pyarrow as pa + + array = cast("Any", column.combine_chunks() if isinstance(column, pa.ChunkedArray) else column) + storage_values = cast("list[Any]", array.storage.to_pylist()) + return [UUID(bytes=value) if value is not None else None for value in storage_values] + if _arrow_type_is_list_of_opaque_uuid(data_type): + nested_values = cast("list[Any]", column.to_pylist()) + return [ + [UUID(bytes=item) if item is not None else None for item in value] if value is not None else None + for value in nested_values + ] + return cast("list[Any]", column.to_pylist()) diff --git a/tests/unit/adapters/test_adbc/test_uuid_decoding.py b/tests/unit/adapters/test_adbc/test_uuid_decoding.py index 6e11b58d8..0497f8180 100644 --- a/tests/unit/adapters/test_adbc/test_uuid_decoding.py +++ b/tests/unit/adapters/test_adbc/test_uuid_decoding.py @@ -47,6 +47,21 @@ def test_plain_arrow_table_uses_whole_table_pylist_path() -> None: assert table.to_pylist_calls == 1 +def test_streamed_opaque_uuid_columns_match_buffered_row_decoding() -> None: + table = _build_uuid_table() + driver = _make_driver(table) + + with driver.select_stream( + "SELECT identifier, identifiers, label FROM uuid_values", native_only=True, chunk_size=1 + ) as stream: + rows = list(stream) + + assert rows == [ + {"identifier": UUID_VALUE, "identifiers": [UUID_VALUE, None, OTHER_UUID_VALUE], "label": "first"}, + {"identifier": None, "identifiers": [], "label": "second"}, + ] + + class _AdbcUuidResultCursor: def __init__(self, table: pa.Table) -> None: self.closed = False @@ -59,6 +74,9 @@ def execute(self, sql: str, parameters: object = None) -> None: def fetch_arrow_table(self) -> pa.Table: return self._table + def fetch_record_batch(self) -> pa.RecordBatchReader: + return pa.RecordBatchReader.from_batches(self._table.schema, self._table.to_batches(max_chunksize=1)) + def close(self) -> None: self.closed = True @@ -107,16 +125,11 @@ def _make_driver(table: pa.Table, *, enable_arrow_extension_types: bool = True) def _build_uuid_table() -> pa.Table: opaque_type = pa.opaque(pa.binary(), "uuid", "PostgreSQL") - scalar_values = pa.ExtensionArray.from_storage( - opaque_type, - pa.array([UUID_VALUE.bytes, None], type=pa.binary()), - ) + scalar_values = pa.ExtensionArray.from_storage(opaque_type, pa.array([UUID_VALUE.bytes, None], type=pa.binary())) list_values = pa.ExtensionArray.from_storage( - opaque_type, - pa.array([UUID_VALUE.bytes, None, OTHER_UUID_VALUE.bytes], type=pa.binary()), + opaque_type, pa.array([UUID_VALUE.bytes, None, OTHER_UUID_VALUE.bytes], type=pa.binary()) ) uuid_lists = pa.ListArray.from_arrays(pa.array([0, 3, 3], type=pa.int32()), list_values) return pa.Table.from_arrays( - [scalar_values, uuid_lists, pa.array(["first", "second"])], - names=["identifier", "identifiers", "label"], + [scalar_values, uuid_lists, pa.array(["first", "second"])], names=["identifier", "identifiers", "label"] ) From 0938adc1c57caaf28d071589d41f35c757b2b2ad Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 17:50:28 +0000 Subject: [PATCH 03/15] test(adbc): verify PostgreSQL UUID round trips --- .../adapters/postgres/adbc/test_driver.py | 29 +++++++------------ 1 file changed, 10 insertions(+), 19 deletions(-) diff --git a/tests/integration/adapters/postgres/adbc/test_driver.py b/tests/integration/adapters/postgres/adbc/test_driver.py index 2eb1429b7..76297c172 100644 --- a/tests/integration/adapters/postgres/adbc/test_driver.py +++ b/tests/integration/adapters/postgres/adbc/test_driver.py @@ -44,10 +44,8 @@ def test_postgresql_uuid_identity_and_same_sql_cache_reuse(postgresql_session: A for position, value in enumerate(values, 1): postgresql_session.execute(insert_sql, (position, value)) - rows = postgresql_session.execute( - f"SELECT position, value::text AS value FROM {table_name} ORDER BY position" - ).get_data() - assert [row["value"] for row in rows] == [str(value) for value in values] + rows = postgresql_session.execute(f"SELECT position, value FROM {table_name} ORDER BY position").get_data() + assert [row["value"] for row in rows] == values finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") @@ -86,10 +84,8 @@ def test_postgresql_uuid_batch_inference(postgresql_session: AdbcDriver) -> None [(1, str(first_value).upper()), (2, None), (3, last_value)], ) - rows = postgresql_session.execute( - f"SELECT position, value::text AS value FROM {table_name} ORDER BY position" - ).get_data() - assert [row["value"] for row in rows] == [str(first_value), None, str(last_value)] + rows = postgresql_session.execute(f"SELECT position, value FROM {table_name} ORDER BY position").get_data() + assert [row["value"] for row in rows] == [first_value, None, last_value] finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") @@ -116,10 +112,10 @@ def test_postgresql_uuid_array_parameter_matches_rows(postgresql_session: AdbcDr postgresql_session.execute(f"INSERT INTO {table_name} (id) VALUES (?)", (value,)) rows = postgresql_session.execute( - f"SELECT id::text AS id FROM {table_name} WHERE {predicate} ORDER BY id::text", (wanted,) + f"SELECT id FROM {table_name} WHERE {predicate} ORDER BY id", (wanted,) ).get_data() - assert sorted(row["id"] for row in rows) == sorted(str(value) for value in wanted) + assert [row["id"] for row in rows] == sorted(wanted) finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") @@ -163,13 +159,10 @@ def test_postgresql_uuid_array_batch_binding(postgresql_session: AdbcDriver) -> ) rows = postgresql_session.execute( - f"SELECT position, array_to_string(identifiers, ',') AS identifiers FROM {table_name} ORDER BY position" + f"SELECT position, identifiers FROM {table_name} ORDER BY position" ).get_data() - assert [row["identifiers"] for row in rows] == [ - ",".join(str(value) for value in first_row), - ",".join(str(value) for value in second_row), - ] + assert [row["identifiers"] for row in rows] == [first_row, second_row] finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") @@ -205,9 +198,7 @@ def test_postgresql_lone_uuid_parameter_survives_statement_cache_hits(postgresql for value in values: postgresql_session.execute(f"INSERT INTO {table_name} (value) VALUES (?)", (value,)) - rows = postgresql_session.execute( - f"SELECT value::text AS value FROM {table_name} ORDER BY value::text" - ).get_data() - assert sorted(row["value"] for row in rows) == sorted(str(value) for value in values) + rows = postgresql_session.execute(f"SELECT value FROM {table_name} ORDER BY value").get_data() + assert [row["value"] for row in rows] == sorted(values) finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") From 649e5cdbefd5cfd01bb45f5d6609d585378d007b Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 18:02:41 +0000 Subject: [PATCH 04/15] perf(adbc): cache UUID schema decisions --- sqlspec/utils/arrow_helpers.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sqlspec/utils/arrow_helpers.py b/sqlspec/utils/arrow_helpers.py index 62593769d..a82235e5f 100644 --- a/sqlspec/utils/arrow_helpers.py +++ b/sqlspec/utils/arrow_helpers.py @@ -518,6 +518,7 @@ def arrow_reader_to_return_format( return shaped, int(table.num_rows) +@lru_cache(maxsize=_ARROW_SCHEMA_DECISION_CACHE_SIZE) def _arrow_schema_has_opaque_uuid(schema: Any) -> bool: return any( _arrow_type_is_opaque_uuid(field.type) or _arrow_type_is_list_of_opaque_uuid(field.type) for field in schema @@ -535,8 +536,7 @@ def _arrow_type_is_list_of_opaque_uuid(data_type: Any) -> bool: ensure_pyarrow() import pyarrow as pa - is_list = pa.types.is_list(data_type) or pa.types.is_large_list(data_type) or pa.types.is_fixed_size_list(data_type) - return is_list and _arrow_type_is_opaque_uuid(data_type.value_type) + return pa.types.is_list(data_type) and _arrow_type_is_opaque_uuid(data_type.value_type) def _arrow_uuid_column_to_pylist(column: Any, data_type: Any) -> "list[Any]": From 61543663ba004de569eabb5f1a8d3ce220a071c7 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 18:23:59 +0000 Subject: [PATCH 05/15] test(adbc): prove UUID row and Arrow boundaries --- .../adapters/duckdb/adbc/test_driver.py | 20 ++++++++++++++++ .../adapters/postgres/adbc/test_driver.py | 23 ++++++++++++++++++- .../adapters/sqlite/adbc/test_driver.py | 17 ++++++++++++++ 3 files changed, 59 insertions(+), 1 deletion(-) diff --git a/tests/integration/adapters/duckdb/adbc/test_driver.py b/tests/integration/adapters/duckdb/adbc/test_driver.py index 943448b77..e7a6e0153 100644 --- a/tests/integration/adapters/duckdb/adbc/test_driver.py +++ b/tests/integration/adapters/duckdb/adbc/test_driver.py @@ -1,6 +1,26 @@ """DuckDB-backed ADBC driver residuals.""" +import pytest + +from sqlspec.adapters.adbc import AdbcDriver from tests.integration.adapters._shared.adbc_backends import duckdb_session, test_duckdb_specific_features from tests.integration.adapters._shared.adbc_connection import test_duckdb_connection __all__ = ("duckdb_session", "test_duckdb_connection", "test_duckdb_specific_features") + + +@pytest.mark.xdist_group("duckdb") +@pytest.mark.adbc +def test_duckdb_uuid_schema_bypasses_opaque_uuid_decoding(duckdb_session: AdbcDriver) -> None: + """DuckDB UUID schemas stay on the ordinary whole-table row path.""" + statement = "SELECT UUID '550e8400-e29b-41d4-a716-446655440000' AS value" + + arrow_table = duckdb_session.select_to_arrow(statement).data + rows = duckdb_session.select(statement) + arrow_type = arrow_table.schema.field("value").type + + assert not ( + getattr(arrow_type, "extension_name", None) == "arrow.opaque" + and getattr(arrow_type, "type_name", None) == "uuid" + ) + assert rows == arrow_table.to_pylist() diff --git a/tests/integration/adapters/postgres/adbc/test_driver.py b/tests/integration/adapters/postgres/adbc/test_driver.py index 76297c172..38e46dfad 100644 --- a/tests/integration/adapters/postgres/adbc/test_driver.py +++ b/tests/integration/adapters/postgres/adbc/test_driver.py @@ -1,6 +1,6 @@ """PostgreSQL-backed ADBC driver residuals.""" -from uuid import uuid4 +from uuid import UUID, uuid4 import pytest @@ -202,3 +202,24 @@ def test_postgresql_lone_uuid_parameter_survives_statement_cache_hits(postgresql assert [row["value"] for row in rows] == sorted(values) finally: postgresql_session.execute_script(f"DROP TABLE IF EXISTS {table_name}") + + +@pytest.mark.xdist_group("postgres") +@pytest.mark.adbc +def test_postgresql_uuid_row_stream_and_arrow_boundary(postgresql_session: AdbcDriver) -> None: + """Row APIs decode UUIDs while the Arrow API preserves the opaque extension.""" + value = UUID("550e8400-e29b-41d4-a716-446655440000") + statement = f"SELECT CAST('{value}' AS UUID) AS value" + + buffered = postgresql_session.select_one(statement) + with postgresql_session.select_stream(statement, native_only=True, chunk_size=1) as stream: + streamed = list(stream) + arrow_result = postgresql_session.select_to_arrow(statement) + arrow_table = arrow_result.data + arrow_type = arrow_table.schema.field("value").type + + assert buffered == {"value": value} + assert streamed == [buffered] + assert arrow_type.extension_name == "arrow.opaque" + assert arrow_type.type_name == "uuid" + assert arrow_table.to_pylist() == [{"value": value.bytes}] diff --git a/tests/integration/adapters/sqlite/adbc/test_driver.py b/tests/integration/adapters/sqlite/adbc/test_driver.py index 8203d9f4f..65e5c1e9c 100644 --- a/tests/integration/adapters/sqlite/adbc/test_driver.py +++ b/tests/integration/adapters/sqlite/adbc/test_driver.py @@ -1,5 +1,8 @@ """SQLite-backed ADBC driver residuals.""" +import pytest + +from sqlspec.adapters.adbc import AdbcDriver from tests.integration.adapters._shared.adbc_backends import sqlite_session, test_sqlite_adbc_specific_features from tests.integration.adapters._shared.adbc_connection import test_sqlite_connection from tests.integration.adapters._shared.adbc_driver import ( @@ -16,3 +19,17 @@ "test_sqlite_adbc_specific_features", "test_sqlite_connection", ) + + +@pytest.mark.xdist_group("sqlite") +@pytest.mark.adbc +def test_sqlite_arrow_schema_bypasses_opaque_uuid_decoding(sqlite_session: AdbcDriver) -> None: + """SQLite Arrow schemas stay on the ordinary whole-table row path.""" + statement = "SELECT 1 AS value" + + arrow_table = sqlite_session.select_to_arrow(statement).data + rows = sqlite_session.select(statement) + arrow_type = arrow_table.schema.field("value").type + + assert getattr(arrow_type, "extension_name", None) != "arrow.opaque" + assert rows == arrow_table.to_pylist() From 42f04064c667bc3d8f05210748a6b97f1a30ae92 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 18:30:27 +0000 Subject: [PATCH 06/15] docs(adbc): explain UUID row decoding --- docs/changelog.rst | 5 ++++ docs/reference/adapters/adbc.rst | 39 ++++++++++++++++++++++++++++---- 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/docs/changelog.rst b/docs/changelog.rst index fcc5c8a97..d4b10baed 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -21,6 +21,11 @@ v0.56.1 **Fixed:** +* PostgreSQL-family ADBC row APIs now decode scalar ``UUID`` and ``UUID[]`` + opaque storage values to Python UUIDs across buffered and streamed results. + Native Arrow results preserve the extension schema, and + ``enable_arrow_extension_types=False`` restores raw storage bytes on row + APIs. * PostgreSQL-family ADBC drivers no longer fail when a UUID is a statement's only parameter. Repeated executions of such a statement previously reused a cached plan that skipped UUID binding and reached PostgreSQL as ``bytea``. diff --git a/docs/reference/adapters/adbc.rst b/docs/reference/adapters/adbc.rst index b7901819a..7301d879f 100644 --- a/docs/reference/adapters/adbc.rst +++ b/docs/reference/adapters/adbc.rst @@ -2,10 +2,10 @@ ADBC ==== -Arrow Database Connectivity adapter providing native Arrow result handling -without conversion overhead. SQLSpec can load the ADBC drivers for PostgreSQL, -SQLite, DuckDB, BigQuery, Snowflake, Flight SQL, and GizmoSQL from one -``AdbcConfig`` surface. +Arrow Database Connectivity adapter providing native Arrow result handling. +Arrow-return APIs preserve native schemas, while row-return APIs materialize +Python values. SQLSpec can load the ADBC drivers for PostgreSQL, SQLite, DuckDB, +BigQuery, Snowflake, Flight SQL, and GizmoSQL from one ``AdbcConfig`` surface. Connection Configuration ======================== @@ -173,6 +173,37 @@ Use ``gizmosql_backend="sqlite"`` only when the target GizmoSQL server was started with SQLite as its database backend. DuckDB remains the default dialect for GizmoSQL. +PostgreSQL UUID Results +======================= + +PostgreSQL-family ADBC drivers expose ``UUID`` columns as Arrow opaque binary +extension values. SQLSpec decodes those values to :class:`uuid.UUID` on row +APIs, including scalar UUIDs, ``UUID[]`` elements, and nulls. The behavior is +consistent for buffered ``select``/``select_one`` calls and +``select_stream``. + +Native Arrow APIs such as ``select_to_arrow`` preserve the opaque extension +schema and its byte-oriented Arrow semantics. SQLite and DuckDB results are not +rewritten because their ADBC schemas do not use PostgreSQL's opaque UUID +extension. + +Set ``enable_arrow_extension_types=False`` in ``driver_features`` to keep raw +storage bytes on row APIs: + +.. code-block:: python + + from sqlspec.adapters.adbc import AdbcConfig + + postgres = AdbcConfig( + connection_config={ + "driver_name": "postgres", + "uri": "postgresql://user:password@localhost:5432/app", + }, + driver_features={"enable_arrow_extension_types": False}, + ) + +The compatibility alias ``arrow_extension_types`` controls the same feature. + PostgreSQL Extension Dialects ============================= From ded6269e5a77c019ae641862fb6c971ddd5ab9c6 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 18:55:40 +0000 Subject: [PATCH 07/15] fix(oracledb): bind JSON through version-aware LOBs --- .github/workflows/ci.yml | 1 + docs/reference/adapters/oracledb.rst | 196 +++++++++++++++ sqlspec/adapters/oracledb/_json_handlers.py | 17 +- sqlspec/adapters/oracledb/_param_types.py | 5 +- sqlspec/adapters/oracledb/adk/store.py | 4 +- sqlspec/adapters/oracledb/config.py | 38 +-- sqlspec/adapters/oracledb/core.py | 145 ++++++++++- sqlspec/adapters/oracledb/data_dictionary.py | 21 ++ sqlspec/adapters/oracledb/driver.py | 26 +- .../oracledb/test_json_storage_modes.py | 109 +++++++++ tests/integration/fixtures/__init__.py | 6 + tests/integration/fixtures/oracle.py | 32 +++ .../test_oracledb/test_json_handlers.py | 9 + .../test_oracledb/test_lob_coercion.py | 231 +++++++++++++++++- tests/unit/test_bench_oracle_scenarios.py | 50 ++++ tools/scripts/bench.py | 90 +++++++ 16 files changed, 927 insertions(+), 53 deletions(-) create mode 100644 tests/integration/adapters/oracle/oracledb/test_json_storage_modes.py create mode 100644 tests/unit/test_bench_oracle_scenarios.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c1e975982..9d9bad585 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -346,6 +346,7 @@ jobs: cockroachdb/cockroach:latest mysql:8.4 mcr.microsoft.com/mssql/server:2022-latest + gvenzl/oracle-xe:18-slim-faststart gvenzl/oracle-free:23-slim-faststart ghcr.io/goccy/bigquery-emulator:latest gcr.io/cloud-spanner-emulator/emulator:latest diff --git a/docs/reference/adapters/oracledb.rst b/docs/reference/adapters/oracledb.rst index 9767607c2..4e1b72a04 100644 --- a/docs/reference/adapters/oracledb.rst +++ b/docs/reference/adapters/oracledb.rst @@ -6,6 +6,151 @@ Sync and async Oracle adapter using `python-oracledb Any: instance: python-oracledb selects its calling convention via ``inspect.signature(handler)``, which succeeds for a partial of a (compiled) function but raises ``ValueError`` for a compiled ``__call__`` object -- the - latter forces the legacy 6-argument call and breaks fetches. + latter forces the older 6-argument call and breaks fetches. """ return partial(_chained_input_handler, inner, fallback) @@ -144,7 +147,7 @@ def chain_output_handler(inner: Any, fallback: "Any | None") -> Any: return partial(_chained_output_handler, inner, fallback) -def _is_json_payload(value: Any) -> bool: +def is_json_payload(value: Any) -> bool: """Return True if the value should be claimed by the JSON input handler. ``dict`` and ``tuple``/``list`` of dicts are claimed. Sequences whose first @@ -167,10 +170,10 @@ def _is_json_payload(value: Any) -> bool: def _input_type_handler(cursor: "Cursor | AsyncCursor", value: Any, arraysize: int) -> Any: """Oracle input type handler for JSON-shaped Python values.""" - if not _is_json_payload(value): + if not is_json_payload(value): return None - server_major = cast("int | None", getattr(cursor.connection, "_sqlspec_oracle_major", None)) + server_major = resolve_oracle_connection_major(cursor.connection) if server_major is None: return cursor.var(DB_TYPE_JSON, arraysize=arraysize) diff --git a/sqlspec/adapters/oracledb/_param_types.py b/sqlspec/adapters/oracledb/_param_types.py index ba757b710..34017e726 100644 --- a/sqlspec/adapters/oracledb/_param_types.py +++ b/sqlspec/adapters/oracledb/_param_types.py @@ -8,7 +8,8 @@ * :class:`OracleClob` — bind as ``DB_TYPE_CLOB`` regardless of length. * :class:`OracleBlob` — bind as ``DB_TYPE_BLOB`` regardless of length. -* :class:`OracleJson` — bind as native JSON; defers to the C1 input handler. +* :class:`OracleJson` — express JSON intent; the input handler selects the + storage type supported by the connected Oracle version. The wrappers themselves perform no validation — type discipline lives at the routing site so error messages can include database-context detail. @@ -38,7 +39,7 @@ def __init__(self, value: "bytes | str") -> None: class OracleJson: - """Mark a value to be bound as native ``DB_TYPE_JSON`` regardless of detected version.""" + """Mark a value for version-aware Oracle JSON binding.""" __slots__ = ("value",) diff --git a/sqlspec/adapters/oracledb/adk/store.py b/sqlspec/adapters/oracledb/adk/store.py index 99bdda2cb..50933c576 100644 --- a/sqlspec/adapters/oracledb/adk/store.py +++ b/sqlspec/adapters/oracledb/adk/store.py @@ -296,7 +296,7 @@ class OracleAsyncADKStore(BaseAsyncADKStore["OracleAsyncConfig"]): config: OracleAsyncConfig with extension_config["adk"] settings. Notes: - - JSON storage type detected based on Oracle version (21c+, 12c+, legacy) + - JSON storage type detected by capability (native JSON, BLOB IS JSON, or plain LOB) - event_data stored as JSON (21c+) or BLOB (older versions) - TIMESTAMP WITH TIME ZONE for timezone-aware timestamps - Named parameters using :param_name @@ -1265,7 +1265,7 @@ class OracleSyncADKStore(BaseSyncADKStore["OracleSyncConfig"]): config: OracleSyncConfig with extension_config["adk"] settings. Notes: - - JSON storage type detected based on Oracle version (21c+, 12c+, legacy) + - JSON storage type detected by capability (native JSON, BLOB IS JSON, or plain LOB) - event_data stored as JSON (21c+) or BLOB (older versions) - TIMESTAMP WITH TIME ZONE for timezone-aware timestamps - Named parameters using :param_name diff --git a/sqlspec/adapters/oracledb/config.py b/sqlspec/adapters/oracledb/config.py index 0e0138484..518a0ad32 100644 --- a/sqlspec/adapters/oracledb/config.py +++ b/sqlspec/adapters/oracledb/config.py @@ -23,7 +23,7 @@ from sqlspec.adapters.oracledb._uuid_handlers import register_uuid_handlers from sqlspec.adapters.oracledb._vector_handlers import register_numpy_handlers # pyright: ignore[reportPrivateUsage] from sqlspec.adapters.oracledb.core import apply_driver_features, default_statement_config -from sqlspec.adapters.oracledb.data_dictionary import OracleVersionCache +from sqlspec.adapters.oracledb.data_dictionary import OracleVersionCache, resolve_oracle_connection_major from sqlspec.adapters.oracledb.driver import ( OracleAsyncDriver, OracleAsyncExceptionHandler, @@ -32,7 +32,6 @@ ) from sqlspec.adapters.oracledb.migrations import OracleAsyncMigrationTracker, OracleSyncMigrationTracker from sqlspec.config import AsyncDatabaseConfig, ExtensionConfigs, SyncDatabaseConfig -from sqlspec.data_dictionary.dialects.oracle import parse_oracle_version_components from sqlspec.driver._async import AsyncPoolConnectionContext, AsyncPoolSessionFactory from sqlspec.driver._sync import SyncPoolConnectionContext, SyncPoolSessionFactory from sqlspec.extensions.events import EventRuntimeHints @@ -413,9 +412,9 @@ def _init_connection(self, connection: "OracleSyncConnection", tag: str) -> None vector case) internally. UUID registration remains gated for backwards compatibility with existing user configurations. - Caches ``connection._sqlspec_oracle_major`` so the JSON input handler - can pick the right binding path (``DB_TYPE_JSON`` on 21c+, OSON-encoded - ``DB_TYPE_BLOB`` on 19c-20c, JSON-string ``DB_TYPE_CLOB`` on 12c-18c) + Caches ``connection._sqlspec_oracle_major`` so JSON parameter coercion + can pick the right binding path (``DB_TYPE_JSON`` on 21c+, a textual + ``BLOB IS JSON`` locator on 12c-20c, and ``DB_TYPE_CLOB`` before 12c) without re-querying server metadata on every bind. Caches ``connection._sqlspec_vector_return_format`` so the vector output handler can dispatch to ``numpy`` / ``list`` / ``array`` without @@ -434,7 +433,9 @@ def _init_connection(self, connection: "OracleSyncConnection", tag: str) -> None # Stash detected major version on the connection so the JSON input handler # can pick the right binding path without per-bind metadata queries. - setattr(connection, "_sqlspec_oracle_major", _resolve_connection_major(self._oracle_version_cache, connection)) + setattr( + connection, "_sqlspec_oracle_major", resolve_oracle_connection_major(connection, self._oracle_version_cache) + ) # Stash the vector-read format so the VECTOR output handler can # dispatch without re-reading driver-feature defaults on every fetch. setattr(connection, "_sqlspec_vector_return_format", self.driver_features.get("vector_return_format")) @@ -628,7 +629,9 @@ async def _init_connection(self, connection: "OracleAsyncConnection", tag: str) # Stash detected major version on the connection so the JSON input handler # can pick the right binding path without per-bind metadata queries. - setattr(connection, "_sqlspec_oracle_major", _resolve_connection_major(self._oracle_version_cache, connection)) + setattr( + connection, "_sqlspec_oracle_major", resolve_oracle_connection_major(connection, self._oracle_version_cache) + ) # Stash the vector-read format so the VECTOR output handler can # dispatch without re-reading driver-feature defaults on every fetch. setattr(connection, "_sqlspec_vector_return_format", self.driver_features.get("vector_return_format")) @@ -655,24 +658,3 @@ async def _close_pool(self) -> None: await self.connection_instance.close() self.connection_instance = None self._oracle_version_cache.reset() - - -def _resolve_connection_major(cache: "OracleVersionCache", connection: Any) -> "int | None": - """Resolve the Oracle server major for connection-setup type handlers. - - Prefers the pool-scoped version cache once a driver has resolved it through - the data-dictionary path. Before that, the major is parsed from - ``connection.version`` — a connection attribute populated at connect time, so - no query is issued — using the same version parser the data dictionary uses. - Returns ``None`` when unavailable; callers treat ``None`` as "assume 21c+". - """ - if cache.resolved and cache.version is not None: - return cache.version.major - try: - version_str = connection.version - except AttributeError: - return None - if not version_str: - return None - components = parse_oracle_version_components(str(version_str)) - return components[0] if components is not None else None diff --git a/sqlspec/adapters/oracledb/core.py b/sqlspec/adapters/oracledb/core.py index ef12a4c09..ca8120a2b 100644 --- a/sqlspec/adapters/oracledb/core.py +++ b/sqlspec/adapters/oracledb/core.py @@ -5,7 +5,10 @@ from collections.abc import Sized from typing import TYPE_CHECKING, Any, Protocol, cast +from sqlspec.adapters.oracledb._json_handlers import is_json_payload from sqlspec.adapters.oracledb._param_types import OracleBlob, OracleClob, OracleJson +from sqlspec.adapters.oracledb._typing import DB_TYPE_BLOB, DB_TYPE_CLOB +from sqlspec.adapters.oracledb.data_dictionary import resolve_oracle_connection_major from sqlspec.adapters.oracledb.type_converter import OracleOutputConverter from sqlspec.core import ( DriverParameterProfile, @@ -15,6 +18,7 @@ build_statement_config_from_profile, create_sql_result, ) +from sqlspec.data_dictionary.dialects.oracle import ORACLE_JSON_STORAGE_BLOB_JSON, resolve_oracle_json_storage from sqlspec.driver import rows_to_dicts from sqlspec.exceptions import ( CheckViolationError, @@ -66,6 +70,8 @@ "build_truncate_statement", "coerce_large_parameters_async", "coerce_large_parameters_sync", + "coerce_many_parameters_async", + "coerce_many_parameters_sync", "collect_async_rows", "collect_sync_rows", "connection_is_thin", @@ -250,7 +256,14 @@ def normalize_execute_many_parameters_async(parameters: Any) -> Any: def coerce_large_parameters_sync( - connection: Any, parameters: Any, *, clob_type: Any, blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, ) -> Any: """Coerce large string/bytes parameters into CLOBs/BLOBs with wrapper-aware routing. @@ -271,6 +284,7 @@ def coerce_large_parameters_sync( blob_type: Oracle BLOB DB type (``oracledb.DB_TYPE_BLOB``). varchar2_byte_limit: Byte-length threshold for implicit CLOB conversion. raw_byte_limit: Byte-length threshold for implicit BLOB conversion. + version_cache: Optional pool-scoped Oracle version cache. Returns: Parameters payload with values routed to the correct LOB / native type. @@ -286,6 +300,7 @@ def coerce_large_parameters_sync( blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, + version_cache=version_cache, ) return parameters if isinstance(parameters, (list, tuple)): @@ -297,6 +312,7 @@ def coerce_large_parameters_sync( blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, + version_cache=version_cache, ) for value in parameters ] @@ -304,7 +320,14 @@ def coerce_large_parameters_sync( async def coerce_large_parameters_async( - connection: Any, parameters: Any, *, clob_type: Any, blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, ) -> Any: """Async mirror of :func:`coerce_large_parameters_sync`. @@ -322,6 +345,7 @@ async def coerce_large_parameters_async( blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, + version_cache=version_cache, ) return parameters if isinstance(parameters, (list, tuple)): @@ -333,12 +357,69 @@ async def coerce_large_parameters_async( blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, + version_cache=version_cache, ) for value in parameters ] return parameters +def coerce_many_parameters_sync( + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, +) -> Any: + """Coerce every parameter row prepared for synchronous ``executemany``.""" + normalized = normalize_execute_many_parameters_sync(parameters) + if not normalized: + return normalized + return [ + coerce_large_parameters_sync( + connection, + row, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + version_cache=version_cache, + ) + for row in normalized + ] + + +async def coerce_many_parameters_async( + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, +) -> Any: + """Coerce every parameter row prepared for asynchronous ``executemany``.""" + normalized = normalize_execute_many_parameters_async(parameters) + if not normalized: + return normalized + return [ + await coerce_large_parameters_async( + connection, + row, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + version_cache=version_cache, + ) + for row in normalized + ] + + def build_insert_statement(table: str, columns: "list[str]") -> str: column_list = ", ".join(columns) placeholders = ", ".join(f":{idx + 1}" for idx in range(len(columns))) @@ -574,7 +655,16 @@ def start(self) -> None: fetch_kwargs = build_fetch_kwargs(self._driver.driver_features) if self._fetch_lobs is not None: fetch_kwargs["fetch_lobs"] = self._fetch_lobs - cast("Any", cursor).execute(self._sql, self._parameters or {}, **fetch_kwargs) + parameters = coerce_large_parameters_sync( + self._driver.connection, + self._parameters, + clob_type=DB_TYPE_CLOB, + blob_type=DB_TYPE_BLOB, + varchar2_byte_limit=self._driver.driver_features.get("oracle_varchar2_byte_limit", 4000), + raw_byte_limit=self._driver.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=getattr(self._driver, "_oracle_version_cache", None), + ) + cast("Any", cursor).execute(self._sql, parameters or {}, **fetch_kwargs) self._driver._check_pending_exception(handler) def fetch_chunk(self) -> "list[dict[str, Any]]": @@ -633,7 +723,16 @@ async def start(self) -> None: fetch_kwargs = build_fetch_kwargs(self._driver.driver_features) if self._fetch_lobs is not None: fetch_kwargs["fetch_lobs"] = self._fetch_lobs - await cast("Any", cursor).execute(self._sql, self._parameters or {}, **fetch_kwargs) + parameters = await coerce_large_parameters_async( + self._driver.connection, + self._parameters, + clob_type=DB_TYPE_CLOB, + blob_type=DB_TYPE_BLOB, + varchar2_byte_limit=self._driver.driver_features.get("oracle_varchar2_byte_limit", 4000), + raw_byte_limit=self._driver.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=getattr(self._driver, "_oracle_version_cache", None), + ) + await cast("Any", cursor).execute(self._sql, parameters or {}, **fetch_kwargs) self._driver._check_pending_exception(handler) async def fetch_chunk(self) -> "list[dict[str, Any]]": @@ -854,7 +953,14 @@ def _resolve_row_metadata(self, description: object) -> tuple[list[str], bool]: def _coerce_value_sync( - connection: Any, value: Any, *, clob_type: Any, blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int + connection: Any, + value: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, ) -> Any: """Route a single parameter value through wrapper-aware coercion (sync).""" if isinstance(value, OracleClob): @@ -868,7 +974,15 @@ def _coerce_value_sync( inner = inner.encode("utf-8") return connection.createlob(blob_type, inner) if isinstance(value, OracleJson): - return value.value + value = value.value + server_major = resolve_oracle_connection_major(connection, version_cache) + if ( + isinstance(server_major, int) + and not isinstance(server_major, bool) + and resolve_oracle_json_storage(server_major) == ORACLE_JSON_STORAGE_BLOB_JSON + and is_json_payload(value) + ): + return connection.createlob(blob_type, to_json(value, as_bytes=True)) if isinstance(value, str) and len(value.encode("utf-8")) > varchar2_byte_limit: return connection.createlob(clob_type, value) if isinstance(value, (bytes, bytearray)) and len(value) > raw_byte_limit: @@ -877,7 +991,14 @@ def _coerce_value_sync( async def _coerce_value_async( - connection: Any, value: Any, *, clob_type: Any, blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int + connection: Any, + value: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + version_cache: Any = None, ) -> Any: """Async mirror of :func:`_coerce_value_sync`.""" if isinstance(value, OracleClob): @@ -891,7 +1012,15 @@ async def _coerce_value_async( inner = inner.encode("utf-8") return await connection.createlob(blob_type, inner) if isinstance(value, OracleJson): - return value.value + value = value.value + server_major = resolve_oracle_connection_major(connection, version_cache) + if ( + isinstance(server_major, int) + and not isinstance(server_major, bool) + and resolve_oracle_json_storage(server_major) == ORACLE_JSON_STORAGE_BLOB_JSON + and is_json_payload(value) + ): + return await connection.createlob(blob_type, to_json(value, as_bytes=True)) if isinstance(value, str) and len(value.encode("utf-8")) > varchar2_byte_limit: return await connection.createlob(clob_type, value) if isinstance(value, (bytes, bytearray)) and len(value) > raw_byte_limit: diff --git a/sqlspec/adapters/oracledb/data_dictionary.py b/sqlspec/adapters/oracledb/data_dictionary.py index 48be241f0..3a46412b3 100644 --- a/sqlspec/adapters/oracledb/data_dictionary.py +++ b/sqlspec/adapters/oracledb/data_dictionary.py @@ -61,6 +61,7 @@ "OracleVersionInfo", "OracledbAsyncDataDictionary", "OracledbSyncDataDictionary", + "resolve_oracle_connection_major", "storage_type_from_version", ) @@ -219,6 +220,26 @@ def storage_type_from_version(version_info: "OracleVersionInfo | None") -> JSONS return _storage_type_from_version(version_info) +def resolve_oracle_connection_major(connection: Any, version_cache: "OracleVersionCache | None" = None) -> "int | None": + """Resolve an Oracle server major without issuing a metadata query. + + Pool-scoped data-dictionary metadata is authoritative when it has already + been resolved. Otherwise, use the connection callback cache and finally + python-oracledb's connection ``version`` string. The final fallback matters + for reacquired pool wrappers that do not retain dynamic attributes. + """ + if version_cache is not None and version_cache.resolved and version_cache.version is not None: + return version_cache.version.major + cached_major = getattr(connection, "_sqlspec_oracle_major", None) + if isinstance(cached_major, int) and not isinstance(cached_major, bool): + return cached_major + version = getattr(connection, "version", None) + if not version: + return None + components = parse_oracle_version_components(str(version)) + return components[0] if components is not None else None + + @mypyc_attr(allow_interpreted_subclasses=True, native_class=False) class OracledbSyncDataDictionary(SyncDataDictionaryBase): """Oracle-specific sync data dictionary.""" diff --git a/sqlspec/adapters/oracledb/driver.py b/sqlspec/adapters/oracledb/driver.py index 49a9f5f69..0cbce23ff 100644 --- a/sqlspec/adapters/oracledb/driver.py +++ b/sqlspec/adapters/oracledb/driver.py @@ -28,6 +28,8 @@ build_truncate_statement, coerce_large_parameters_async, coerce_large_parameters_sync, + coerce_many_parameters_async, + coerce_many_parameters_sync, collect_async_rows, collect_sync_rows, connection_is_thin, @@ -35,8 +37,6 @@ default_statement_config, driver_profile, normalize_column_names, - normalize_execute_many_parameters_async, - normalize_execute_many_parameters_sync, resolve_row_metadata, resolve_rowcount, supports_df_batches, @@ -376,6 +376,7 @@ def dispatch_execute(self, cursor: Any, statement: "SQL") -> "ExecutionResult": blob_type=DB_TYPE_BLOB, varchar2_byte_limit=self.driver_features.get("oracle_varchar2_byte_limit", 4000), raw_byte_limit=self.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=self._oracle_version_cache, ) prepared_parameters = cast("list[Any] | tuple[Any, ...] | dict[Any, Any] | None", prepared_parameters) @@ -422,7 +423,15 @@ def dispatch_execute_many(self, cursor: Any, statement: "SQL") -> "ExecutionResu """ sql, prepared_parameters = self._compiled_sql(statement, self.statement_config) - prepared_parameters = normalize_execute_many_parameters_sync(prepared_parameters) + prepared_parameters = coerce_many_parameters_sync( + self.connection, + prepared_parameters, + clob_type=DB_TYPE_CLOB, + blob_type=DB_TYPE_BLOB, + varchar2_byte_limit=self.driver_features.get("oracle_varchar2_byte_limit", 4000), + raw_byte_limit=self.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=self._oracle_version_cache, + ) execution_args = statement.statement_config.execution_args or {} batch_errors = bool(execution_args.get("oracle_batch_errors", False)) array_dml_row_counts = bool(execution_args.get("oracle_array_dml_row_counts", False)) @@ -1080,6 +1089,7 @@ async def dispatch_execute(self, cursor: Any, statement: "SQL") -> "ExecutionRes blob_type=DB_TYPE_BLOB, varchar2_byte_limit=self.driver_features.get("oracle_varchar2_byte_limit", 4000), raw_byte_limit=self.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=self._oracle_version_cache, ) prepared_parameters = cast("list[Any] | tuple[Any, ...] | dict[Any, Any] | None", prepared_parameters) @@ -1126,7 +1136,15 @@ async def dispatch_execute_many(self, cursor: Any, statement: "SQL") -> "Executi """ sql, prepared_parameters = self._compiled_sql(statement, self.statement_config) - prepared_parameters = normalize_execute_many_parameters_async(prepared_parameters) + prepared_parameters = await coerce_many_parameters_async( + self.connection, + prepared_parameters, + clob_type=DB_TYPE_CLOB, + blob_type=DB_TYPE_BLOB, + varchar2_byte_limit=self.driver_features.get("oracle_varchar2_byte_limit", 4000), + raw_byte_limit=self.driver_features.get("oracle_raw_byte_limit", 2000), + version_cache=self._oracle_version_cache, + ) execution_args = statement.statement_config.execution_args or {} batch_errors = bool(execution_args.get("oracle_batch_errors", False)) array_dml_row_counts = bool(execution_args.get("oracle_array_dml_row_counts", False)) diff --git a/tests/integration/adapters/oracle/oracledb/test_json_storage_modes.py b/tests/integration/adapters/oracle/oracledb/test_json_storage_modes.py new file mode 100644 index 000000000..c020c2beb --- /dev/null +++ b/tests/integration/adapters/oracle/oracledb/test_json_storage_modes.py @@ -0,0 +1,109 @@ +"""Oracle JSON storage compatibility coverage.""" + +from decimal import Decimal + +import pytest + +from sqlspec.adapters.oracledb import OracleAsyncDriver, OracleClob, OracleJson +from sqlspec.utils.serializers import to_json + +pytestmark = pytest.mark.xdist_group("oracle") + + +async def _drop_table(driver: "OracleAsyncDriver", table_name: str) -> None: + await driver.execute_script( + f"BEGIN EXECUTE IMMEDIATE 'DROP TABLE {table_name}'; " + "EXCEPTION WHEN OTHERS THEN IF SQLCODE != -942 THEN RAISE; END IF; END;" + ) + + +async def test_non_native_json_session_uses_oracle_18c(oracle_18c_async_session: "OracleAsyncDriver") -> None: + """The non-native JSON lane must run against the real Oracle 18c service.""" + major = await oracle_18c_async_session.select_value( + "SELECT TO_NUMBER(REGEXP_SUBSTR(version, '^[0-9]+')) FROM product_component_version " + "WHERE product LIKE 'Oracle Database%'" + ) + + assert major == 18 + + +async def test_native_json_round_trip_matrix(oracle_async_session: "OracleAsyncDriver") -> None: + """Native JSON accepts direct, wrapped, sequence, and large Python payloads.""" + table_name = "json_modes_native" + payloads: list[object] = [ + {"storage": "native", "number": 3.25}, + [{"item": 1}, {"item": 2}], + {"large": "x" * 5000}, + OracleJson({"explicit": True}), + ] + await _drop_table(oracle_async_session, table_name) + await oracle_async_session.execute_script(f"CREATE TABLE {table_name} (id NUMBER PRIMARY KEY, payload JSON)") + + try: + for row_id, payload in enumerate(payloads, start=1): + await oracle_async_session.execute( + f"INSERT INTO {table_name} (id, payload) VALUES (:id, :payload)", {"id": row_id, "payload": payload} + ) + + rows = await oracle_async_session.select(f"SELECT id, payload FROM {table_name} ORDER BY id") + + assert rows[0]["payload"]["storage"] == "native" + assert isinstance(rows[0]["payload"]["number"], Decimal) + assert rows[1]["payload"] == [{"item": 1}, {"item": 2}] + assert rows[2]["payload"]["large"] == "x" * 5000 + assert rows[3]["payload"] == {"explicit": True} + finally: + await _drop_table(oracle_async_session, table_name) + + +async def test_blob_is_json_round_trip_on_oracle_18c(oracle_18c_async_session: "OracleAsyncDriver") -> None: + """Oracle 18c routes single and array-DML Python JSON through BLOB IS JSON storage.""" + table_name = "json_modes_blob" + direct_payload = {"storage": "blob", "number": 3.25, "large": "x" * 5000} + list_payload = [{"storage": "blob-list", "item": 1}, {"storage": "blob-list", "item": 2}] + wrapped_payload = {"storage": "blob-wrapped", "explicit": True} + await _drop_table(oracle_18c_async_session, table_name) + await oracle_18c_async_session.execute_script( + f"CREATE TABLE {table_name} (id NUMBER PRIMARY KEY, payload BLOB CHECK (payload IS JSON))" + ) + + try: + await oracle_18c_async_session.execute( + f"INSERT INTO {table_name} (id, payload) VALUES (:id, :payload)", {"id": 1, "payload": direct_payload} + ) + await oracle_18c_async_session.execute_many( + f"INSERT INTO {table_name} (id, payload) VALUES (:id, :payload)", + [{"id": 2, "payload": list_payload}, {"id": 3, "payload": OracleJson(wrapped_payload)}], + ) + + rows = await oracle_18c_async_session.select(f"SELECT id, payload FROM {table_name} ORDER BY id") + + assert rows[0]["payload"] == direct_payload + assert isinstance(rows[0]["payload"]["number"], float) + assert rows[1]["payload"] == list_payload + assert rows[2]["payload"] == wrapped_payload + finally: + await _drop_table(oracle_18c_async_session, table_name) + + +async def test_clob_is_json_round_trip_on_oracle_18c(oracle_18c_async_session: "OracleAsyncDriver") -> None: + """Explicit CLOB storage still receives metadata-driven JSON decoding.""" + table_name = "json_modes_clob" + payload = {"storage": "clob", "number": 3.25, "large": "x" * 5000} + await _drop_table(oracle_18c_async_session, table_name) + await oracle_18c_async_session.execute_script( + f"CREATE TABLE {table_name} (id NUMBER PRIMARY KEY, payload CLOB CHECK (payload IS JSON))" + ) + + try: + await oracle_18c_async_session.execute( + f"INSERT INTO {table_name} (id, payload) VALUES (:id, :payload)", + {"id": 1, "payload": OracleClob(to_json(payload))}, + ) + + row = await oracle_18c_async_session.select_one(f"SELECT payload FROM {table_name} WHERE id = :id", {"id": 1}) + + assert row["payload"] == payload + assert isinstance(row["payload"]["number"], float) + finally: + await _drop_table(oracle_18c_async_session, table_name) diff --git a/tests/integration/fixtures/__init__.py b/tests/integration/fixtures/__init__.py index 0261e6f92..226d37852 100644 --- a/tests/integration/fixtures/__init__.py +++ b/tests/integration/fixtures/__init__.py @@ -51,6 +51,9 @@ pymysql_transaction_config, ) from tests.integration.fixtures.oracle import ( + oracle_18c_async_config, + oracle_18c_async_session, + oracle_18c_connection_config, oracle_aq_privileges, oracle_async_config, oracle_async_session, @@ -151,6 +154,9 @@ "mysqlconnector_sync_driver", "mysqlconnector_sync_transaction_config", "native_bigquery_service", + "oracle_18c_async_config", + "oracle_18c_async_session", + "oracle_18c_connection_config", "oracle_aq_privileges", "oracle_async_config", "oracle_async_session", diff --git a/tests/integration/fixtures/oracle.py b/tests/integration/fixtures/oracle.py index 3a685904b..35d141e25 100644 --- a/tests/integration/fixtures/oracle.py +++ b/tests/integration/fixtures/oracle.py @@ -14,6 +14,9 @@ ) __all__ = ( + "oracle_18c_async_config", + "oracle_18c_async_session", + "oracle_18c_connection_config", "oracle_aq_privileges", "oracle_async_config", "oracle_async_session", @@ -35,6 +38,35 @@ def _oracle_pool_params(oracle_service: "OracleService") -> "OraclePoolParams": ) +@pytest.fixture(scope="session") +def oracle_18c_connection_config(oracle_18c_service: "OracleService") -> "OraclePoolParams": + """Provide Oracle 18c pool parameters for non-native JSON storage coverage.""" + return _oracle_pool_params(oracle_18c_service) + + +@pytest.fixture(scope="session") +async def oracle_18c_async_config( + oracle_18c_connection_config: "OraclePoolParams", +) -> "AsyncGenerator[OracleAsyncConfig, None]": + """Provide a session-scoped Oracle 18c async configuration.""" + config = OracleAsyncConfig(connection_config=OraclePoolParams(**oracle_18c_connection_config)) + try: + yield config + finally: + if config.connection_instance is not None: + await config.close_pool() + config.connection_instance = None + + +@pytest.fixture +async def oracle_18c_async_session( + oracle_18c_async_config: "OracleAsyncConfig", +) -> "AsyncGenerator[OracleAsyncDriver, None]": + """Create an Oracle 18c async driver session.""" + async with oracle_18c_async_config.provide_session() as driver: + yield driver + + @pytest.fixture(scope="session") def oracle_connection_config(oracle_23ai_service: "OracleService") -> "OraclePoolParams": """Provide shared Oracle pool parameters.""" diff --git a/tests/unit/adapters/test_oracledb/test_json_handlers.py b/tests/unit/adapters/test_oracledb/test_json_handlers.py index 39a78fcc0..67d3fe60b 100644 --- a/tests/unit/adapters/test_oracledb/test_json_handlers.py +++ b/tests/unit/adapters/test_oracledb/test_json_handlers.py @@ -16,6 +16,7 @@ json_output_type_handler, register_json_handlers, ) +from sqlspec.adapters.oracledb.data_dictionary import resolve_oracle_connection_major def _mock_cursor_with_major(major: int | None) -> Mock: @@ -115,6 +116,14 @@ def test_input_handler_routes_dict_to_db_type_json_on_21c_plus() -> None: cursor.var.assert_called_once_with(oracledb.DB_TYPE_JSON, arraysize=5) +def test_resolve_oracle_connection_major_falls_back_to_connection_version() -> None: + """Pool wrappers without the cached attribute still expose the server version.""" + connection = Mock(spec=["version"]) + connection.version = "18.0.0.0.0" + + assert resolve_oracle_connection_major(connection) == 18 + + def test_input_handler_routes_dict_to_db_type_blob_on_19c() -> None: """Dict bound on Oracle 19c-20c should use DB_TYPE_BLOB with textual-JSON inconverter.""" import oracledb diff --git a/tests/unit/adapters/test_oracledb/test_lob_coercion.py b/tests/unit/adapters/test_oracledb/test_lob_coercion.py index 7631ef3b6..92c708410 100644 --- a/tests/unit/adapters/test_oracledb/test_lob_coercion.py +++ b/tests/unit/adapters/test_oracledb/test_lob_coercion.py @@ -1,10 +1,24 @@ -"""Unit tests for Oracle LOB parameter coercion.""" +"""Unit tests for Oracle LOB and JSON parameter coercion.""" +from collections.abc import Callable +from typing import cast from unittest.mock import AsyncMock, MagicMock import pytest -from sqlspec.adapters.oracledb.core import coerce_large_parameters_async, coerce_large_parameters_sync +from sqlspec.adapters.oracledb import OracleJson +from sqlspec.adapters.oracledb._typing import OracleAsyncConnection, OracleSyncConnection +from sqlspec.adapters.oracledb.core import ( + OracleAsyncStreamSource, + OracleSyncStreamSource, + coerce_large_parameters_async, + coerce_large_parameters_sync, + coerce_many_parameters_async, + coerce_many_parameters_sync, +) +from sqlspec.adapters.oracledb.data_dictionary import OracleVersionCache, OracleVersionInfo +from sqlspec.adapters.oracledb.driver import OracleAsyncDriver, OracleSyncDriver +from sqlspec.utils.serializers import to_json CLOB_TYPE = "DB_TYPE_CLOB" BLOB_TYPE = "DB_TYPE_BLOB" @@ -12,6 +26,14 @@ RAW_LIMIT = 2000 +def _direct_json(payload: object) -> object: + return payload + + +def _wrapped_json(payload: object) -> OracleJson: + return OracleJson(payload) + + @pytest.fixture def sync_connection() -> MagicMock: conn = MagicMock() @@ -26,6 +48,211 @@ def async_connection() -> AsyncMock: return conn +@pytest.mark.parametrize("major", [12, 18, 20]) +@pytest.mark.parametrize( + ("payload", "wrapper"), + [({"kind": "mapping"}, _direct_json), ([{"kind": "sequence"}], _direct_json), ({"kind": "wrapped"}, _wrapped_json)], + ids=["dict", "list", "oracle-json"], +) +def test_coerce_json_parameters_sync_pre_native_versions_create_utf8_blob_locator( + sync_connection: MagicMock, major: int, payload: object, wrapper: "Callable[[object], object]" +) -> None: + """Oracle 12c-20c JSON values become textual UTF-8 BLOB locators before binding.""" + sync_connection._sqlspec_oracle_major = major + value = wrapper(payload) + + result = coerce_large_parameters_sync( + sync_connection, + {"payload": value}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + sync_connection.createlob.assert_called_once_with(BLOB_TYPE, to_json(payload, as_bytes=True)) + assert result["payload"] is sync_connection.createlob.return_value + + +@pytest.mark.parametrize( + ("payload", "wrapper"), + [({"kind": "mapping"}, _direct_json), ([{"kind": "sequence"}], _direct_json), ({"kind": "wrapped"}, _wrapped_json)], + ids=["dict", "list", "oracle-json"], +) +def test_coerce_json_parameters_sync_native_versions_keep_python_value_for_db_type_json( + sync_connection: MagicMock, payload: object, wrapper: "Callable[[object], object]" +) -> None: + """Oracle 21c+ values stay as Python JSON for the DB_TYPE_JSON input handler.""" + sync_connection._sqlspec_oracle_major = 21 + + result = coerce_large_parameters_sync( + sync_connection, + {"payload": wrapper(payload)}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + sync_connection.createlob.assert_not_called() + assert result["payload"] is payload + + +def test_coerce_json_parameters_sync_uses_connection_version_when_cached_major_is_missing( + sync_connection: MagicMock, +) -> None: + """A reacquired Oracle 18c connection still takes the BLOB locator path.""" + payload = {"kind": "reacquired"} + del sync_connection._sqlspec_oracle_major + sync_connection.version = "18.0.0.0.0" + + result = coerce_large_parameters_sync( + sync_connection, + {"payload": payload}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + sync_connection.createlob.assert_called_once_with(BLOB_TYPE, to_json(payload, as_bytes=True)) + assert result["payload"] is sync_connection.createlob.return_value + + +def test_coerce_json_parameters_sync_prefers_pool_scoped_version_cache(sync_connection: MagicMock) -> None: + """Resolved config metadata avoids wrapper attributes and version parsing.""" + payload = {"kind": "cached"} + version_cache = OracleVersionCache() + version_cache.resolved = True + version_cache.version = OracleVersionInfo(18) + del sync_connection._sqlspec_oracle_major + sync_connection.version = "23.0.0.0.0" + + result = coerce_large_parameters_sync( + sync_connection, + {"payload": payload}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + version_cache=version_cache, + ) + + sync_connection.createlob.assert_called_once_with(BLOB_TYPE, to_json(payload, as_bytes=True)) + assert result["payload"] is sync_connection.createlob.return_value + + +def test_coerce_json_parameters_sync_explicit_clob_remains_clob_on_oracle_18c(sync_connection: MagicMock) -> None: + """OracleClob intent takes precedence over version-aware JSON coercion.""" + from sqlspec.adapters.oracledb import OracleClob + + sync_connection._sqlspec_oracle_major = 18 + + result = coerce_large_parameters_sync( + sync_connection, + {"payload": OracleClob('{"kind":"clob"}')}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + sync_connection.createlob.assert_called_once_with(CLOB_TYPE, '{"kind":"clob"}') + assert result["payload"] is sync_connection.createlob.return_value + + +@pytest.mark.anyio +async def test_coerce_json_parameters_async_oracle_18c_creates_utf8_blob_locator(async_connection: AsyncMock) -> None: + """The async path awaits creation of the same textual JSON BLOB locator.""" + payload = {"kind": "async"} + async_connection._sqlspec_oracle_major = 18 + + result = await coerce_large_parameters_async( + async_connection, + {"payload": payload}, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + async_connection.createlob.assert_awaited_once_with(BLOB_TYPE, to_json(payload, as_bytes=True)) + assert result["payload"] is async_connection.createlob.return_value + + +def test_coerce_many_parameters_sync_applies_json_coercion_to_every_row(sync_connection: MagicMock) -> None: + """Executemany coercion visits each named-bind row.""" + sync_connection._sqlspec_oracle_major = 18 + rows = [{"id": 1, "payload": {"row": 1}}, {"id": 2, "payload": [{"row": 2}]}] + + result = coerce_many_parameters_sync( + sync_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + assert sync_connection.createlob.call_count == 2 + assert all(row["payload"] is sync_connection.createlob.return_value for row in result) + + +@pytest.mark.anyio +async def test_coerce_many_parameters_async_applies_json_coercion_to_every_row(async_connection: AsyncMock) -> None: + """Async executemany coercion awaits locator creation for every row.""" + async_connection._sqlspec_oracle_major = 18 + rows = [{"id": 1, "payload": {"row": 1}}, {"id": 2, "payload": [{"row": 2}]}] + + result = await coerce_many_parameters_async( + async_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + assert async_connection.createlob.await_count == 2 + assert all(row["payload"] is async_connection.createlob.return_value for row in result) + + +def test_oracle_sync_stream_source_coerces_json_filter_before_execute() -> None: + """Native sync streaming uses the same Oracle 18c BLOB locator bind path.""" + locator = MagicMock(name="BLOB") + cursor = MagicMock() + connection = MagicMock() + connection._sqlspec_oracle_major = 18 + connection.createlob.return_value = locator + connection.cursor.return_value = cursor + driver = OracleSyncDriver(cast("OracleSyncConnection", connection)) + source = OracleSyncStreamSource(driver, "SELECT :payload FROM dual", {"payload": {"kind": "stream"}}, 100) + + source.start() + + parameters = cursor.execute.call_args.args[1] + assert parameters["payload"] is locator + + +@pytest.mark.anyio +async def test_oracle_async_stream_source_coerces_json_filter_before_execute() -> None: + """Native async streaming awaits the same Oracle 18c BLOB locator bind path.""" + locator = MagicMock(name="BLOB") + cursor = MagicMock() + cursor.execute = AsyncMock() + connection = MagicMock() + connection._sqlspec_oracle_major = 18 + connection.createlob = AsyncMock(return_value=locator) + connection.cursor.return_value = cursor + driver = OracleAsyncDriver(cast("OracleAsyncConnection", connection)) + source = OracleAsyncStreamSource(driver, "SELECT :payload FROM dual", {"payload": {"kind": "stream"}}, 100) + + await source.start() + + parameters = cursor.execute.call_args.args[1] + assert parameters["payload"] is locator + + def test_coerce_large_parameters_sync_none_parameters_passthrough(sync_connection: MagicMock) -> None: result = coerce_large_parameters_sync( sync_connection, diff --git a/tests/unit/test_bench_oracle_scenarios.py b/tests/unit/test_bench_oracle_scenarios.py new file mode 100644 index 000000000..640b76230 --- /dev/null +++ b/tests/unit/test_bench_oracle_scenarios.py @@ -0,0 +1,50 @@ +"""Unit coverage for Oracle JSON benchmark scenario registration.""" + +from collections.abc import Callable + +import pytest +from tools.scripts import bench + + +def test_oracle_json_scenarios_are_registered() -> None: + """Every Oracle JSON benchmark resolves to its public callable.""" + expected: dict[tuple[str, str, str], Callable[[], None]] = { + ("sqlspec_native_json", "oracle", "json_write"): bench.sqlspec_oracle_native_json_write, + ("sqlspec_serialized_json", "oracle", "json_write"): bench.sqlspec_oracle_serialized_json_write, + ("sqlspec", "oracle", "json_read"): bench.sqlspec_oracle_json_read, + } + + for key, scenario in expected.items(): + assert bench.SCENARIO_REGISTRY[key] is scenario + assert (key[0], key[2]) in bench.ORACLE_EXTENDED_SCENARIOS + + +def test_oracle_json_rows_distinguish_native_and_serialized_payloads() -> None: + """Native and serialized writes use equivalent payloads with distinct bind types.""" + native_rows = bench._oracle_json_rows(serialized=False) + serialized_rows = bench._oracle_json_rows(serialized=True) + + assert len(native_rows) == len(serialized_rows) == bench.ORACLE_JSON_ROWS + assert isinstance(native_rows[0][1], dict) + assert isinstance(serialized_rows[0][1], str) + assert native_rows[0][1] == bench.ORACLE_JSON_PAYLOAD + + +def test_oracle_json_public_wrappers_delegate(monkeypatch: "pytest.MonkeyPatch") -> None: + """Public scenarios select the intended write/read runner mode.""" + calls: list[tuple[str, bool | None]] = [] + + def fake_write(*, serialized: bool) -> None: + calls.append(("write", serialized)) + + def fake_read() -> None: + calls.append(("read", None)) + + monkeypatch.setattr(bench, "_run_sqlspec_oracle_json_write", fake_write) + monkeypatch.setattr(bench, "_run_sqlspec_oracle_json_read", fake_read) + + bench.sqlspec_oracle_native_json_write() + bench.sqlspec_oracle_serialized_json_write() + bench.sqlspec_oracle_json_read() + + assert calls == [("write", False), ("write", True), ("read", None)] diff --git a/tools/scripts/bench.py b/tools/scripts/bench.py index b0480cedc..368f0308b 100644 --- a/tools/scripts/bench.py +++ b/tools/scripts/bench.py @@ -103,6 +103,7 @@ "sqlspec_duckdb_repeated_queries", "sqlspec_duckdb_write_heavy", "sqlspec_mysqlconnector_json_rows", + "sqlspec_oracle_json_read", "sqlspec_oracle_lob_fetch_1k", "sqlspec_oracle_lob_fetch_100k", "sqlspec_oracle_lob_fetch_async_1k", @@ -111,6 +112,8 @@ "sqlspec_oracle_lob_fetch_async_fetch_lobs_true_100k", "sqlspec_oracle_lob_fetch_fetch_lobs_true_1k", "sqlspec_oracle_lob_fetch_fetch_lobs_true_100k", + "sqlspec_oracle_native_json_write", + "sqlspec_oracle_serialized_json_write", "sqlspec_psycopg_async_rows", "sqlspec_psycopg_sync_rows", "sqlspec_spanner_strings", @@ -160,6 +163,12 @@ def _is_compiled() -> bool: CORE_SCENARIOS = ("initialization", "write_heavy", "read_heavy", "iterative_inserts", "repeated_queries") ORACLE_LOB_ROWS = 100 ORACLE_LOB_PAYLOAD_SIZES = {"1k": 1024, "100k": 100 * 1024} +ORACLE_JSON_ROWS = 100 +ORACLE_JSON_PAYLOAD = { + "active": True, + "count": 3, + "items": [{"name": "alpha", "value": 1.25}, {"name": "beta", "value": 2.5}], +} ORACLE_LOB_ENV_VARS = ( "SQLSPEC_BENCH_ORACLE_HOST", "SQLSPEC_BENCH_ORACLE_PORT", @@ -182,6 +191,9 @@ def _is_compiled() -> bool: ("sqlspec", "schema_type_numpy"), ) ORACLE_EXTENDED_SCENARIOS = ( + ("sqlspec_native_json", "json_write"), + ("sqlspec_serialized_json", "json_write"), + ("sqlspec", "json_read"), ("raw", "lob_fetch_1k"), ("sqlspec", "lob_fetch_1k"), ("sqlspec_fetch_lobs_true", "lob_fetch_1k"), @@ -2577,6 +2589,7 @@ def sqlspec_sqlite_thin_path_stress() -> None: # --- Oracle LOB fetch scenarios --- ORACLE_LOB_TABLES = {"1k": "SQLSPEC_LOB_1K", "100k": "SQLSPEC_LOB_100K"} +ORACLE_JSON_TABLE = "SQLSPEC_JSON_BENCH" def _get_oracledb() -> Any: @@ -2648,6 +2661,23 @@ def _oracle_lob_rows(size_key: str) -> list[tuple[int, str]]: return [(index, payload) for index in range(1, ORACLE_LOB_ROWS + 1)] +def _oracle_create_json_table_sql() -> str: + return f"CREATE TABLE {ORACLE_JSON_TABLE} (id NUMBER PRIMARY KEY, payload JSON)" + + +def _oracle_insert_json_sql() -> str: + return f"INSERT INTO {ORACLE_JSON_TABLE} (id, payload) VALUES (:1, :2)" + + +def _oracle_select_json_sql() -> str: + return f"SELECT id, payload FROM {ORACLE_JSON_TABLE} ORDER BY id" + + +def _oracle_json_rows(*, serialized: bool) -> list[tuple[int, object]]: + payload: object = json.dumps(ORACLE_JSON_PAYLOAD) if serialized else ORACLE_JSON_PAYLOAD + return [(index, payload) for index in range(1, ORACLE_JSON_ROWS + 1)] + + def _read_oracle_lob_value(value: Any) -> Any: read = getattr(value, "read", None) if callable(read): @@ -2744,6 +2774,48 @@ async def _run_sqlspec_oracle_lob_fetch_async(size_key: str, *, fetch_lobs: bool await config.close_pool() +def _run_sqlspec_oracle_json_write(*, serialized: bool) -> None: + from sqlspec.adapters.oracledb import OracleSyncConfig + + spec = SQLSpec() + config = OracleSyncConfig(connection_config=_oracle_connection_config_from_env()) + try: + with spec.provide_session(config) as session: + session.execute_script(_oracle_drop_table_sql(ORACLE_JSON_TABLE)) + session.execute(_oracle_create_json_table_sql()) + try: + session.execute_many(_oracle_insert_json_sql(), _oracle_json_rows(serialized=serialized)) + finally: + session.execute_script(_oracle_drop_table_sql(ORACLE_JSON_TABLE)) + _check_pool_leak(config.connection_instance, f"oracle/json_write/serialized={serialized}") + config.close_pool() + finally: + if config.connection_instance is not None: + config.close_pool() + + +def _run_sqlspec_oracle_json_read() -> None: + from sqlspec.adapters.oracledb import OracleSyncConfig + + spec = SQLSpec() + config = OracleSyncConfig(connection_config=_oracle_connection_config_from_env()) + try: + with spec.provide_session(config) as session: + session.execute_script(_oracle_drop_table_sql(ORACLE_JSON_TABLE)) + session.execute(_oracle_create_json_table_sql()) + try: + session.execute_many(_oracle_insert_json_sql(), _oracle_json_rows(serialized=False)) + rows = session.fetch(_oracle_select_json_sql()) + assert len(rows) == ORACLE_JSON_ROWS + finally: + session.execute_script(_oracle_drop_table_sql(ORACLE_JSON_TABLE)) + _check_pool_leak(config.connection_instance, "oracle/json_read") + config.close_pool() + finally: + if config.connection_instance is not None: + config.close_pool() + + def raw_oracle_lob_fetch_1k() -> None: """Fetch 100 1 KiB Oracle CLOB rows through raw python-oracledb.""" _run_raw_oracle_lob_fetch("1k") @@ -2794,6 +2866,21 @@ async def sqlspec_oracle_lob_fetch_async_fetch_lobs_true_100k() -> None: await _run_sqlspec_oracle_lob_fetch_async("100k", fetch_lobs=True) +def sqlspec_oracle_native_json_write() -> None: + """Write native Python mappings to an Oracle native JSON column.""" + _run_sqlspec_oracle_json_write(serialized=False) + + +def sqlspec_oracle_serialized_json_write() -> None: + """Write pre-serialized strings to an Oracle native JSON column.""" + _run_sqlspec_oracle_json_write(serialized=True) + + +def sqlspec_oracle_json_read() -> None: + """Read native JSON values from Oracle through SQLSpec.""" + _run_sqlspec_oracle_json_read() + + SCENARIO_REGISTRY: dict[tuple[str, str, str], Any] = { # SQLite scenarios ("raw", "sqlite", "initialization"): raw_sqlite_initialization, @@ -2904,6 +2991,9 @@ async def sqlspec_oracle_lob_fetch_async_fetch_lobs_true_100k() -> None: ("sqlspec_async", "oracle", "lob_fetch_100k"): sqlspec_oracle_lob_fetch_async_100k, ("sqlspec_async_fetch_lobs_true", "oracle", "lob_fetch_1k"): sqlspec_oracle_lob_fetch_async_fetch_lobs_true_1k, ("sqlspec_async_fetch_lobs_true", "oracle", "lob_fetch_100k"): sqlspec_oracle_lob_fetch_async_fetch_lobs_true_100k, + ("sqlspec_native_json", "oracle", "json_write"): sqlspec_oracle_native_json_write, + ("sqlspec_serialized_json", "oracle", "json_write"): sqlspec_oracle_serialized_json_write, + ("sqlspec", "oracle", "json_read"): sqlspec_oracle_json_read, } From 7e7f067c34b731cbc6c2080ce87b6e2dbcb5e192 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 19:13:09 +0000 Subject: [PATCH 08/15] refactor(utils): centralize UUID construction --- sqlspec/adapters/adbc/core.py | 3 +- sqlspec/adapters/duckdb/core.py | 4 +- sqlspec/adapters/oracledb/_uuid_handlers.py | 3 +- sqlspec/adapters/psqlpy/core.py | 3 +- sqlspec/adapters/spanner/type_converter.py | 5 +- sqlspec/core/type_converter.py | 4 +- sqlspec/utils/arrow_helpers.py | 6 +- sqlspec/utils/schema.py | 5 +- sqlspec/utils/type_converters.py | 3 +- sqlspec/utils/uuids.py | 53 ++++++++++++- .../test_oracledb/test_uuid_handlers.py | 4 +- tests/unit/utils/test_uuids.py | 76 +++++++++++++++++++ 12 files changed, 150 insertions(+), 19 deletions(-) diff --git a/sqlspec/adapters/adbc/core.py b/sqlspec/adapters/adbc/core.py index aeac41acd..ae386eb48 100644 --- a/sqlspec/adapters/adbc/core.py +++ b/sqlspec/adapters/adbc/core.py @@ -46,6 +46,7 @@ from sqlspec.utils.serializers import to_json from sqlspec.utils.type_converters import build_uuid_coercions from sqlspec.utils.type_guards import has_rowcount, has_sqlstate +from sqlspec.utils.uuids import uuid_from_string if TYPE_CHECKING: from collections.abc import Callable, Mapping @@ -1068,7 +1069,7 @@ def _convert_uuid_value(value: Any, ordinal: int, row_number: int, element_index if not isinstance(value, str): raise SQLSpecError(_uuid_binding_error(ordinal, value, row_number, element_index)) try: - return str(UUID(value)) + return str(uuid_from_string(value)) except (TypeError, ValueError) as exc: raise SQLSpecError(_uuid_binding_error(ordinal, value, row_number, element_index)) from exc diff --git a/sqlspec/adapters/duckdb/core.py b/sqlspec/adapters/duckdb/core.py index 5a6ce57c7..35183a3c5 100644 --- a/sqlspec/adapters/duckdb/core.py +++ b/sqlspec/adapters/duckdb/core.py @@ -4,7 +4,6 @@ from datetime import date, datetime from decimal import Decimal from typing import TYPE_CHECKING, Any, Final, cast -from uuid import UUID from sqlspec.core import DriverParameterProfile, ParameterStyle, StatementConfig, build_statement_config_from_profile from sqlspec.exceptions import ( @@ -24,6 +23,7 @@ from sqlspec.utils.serializers import to_json from sqlspec.utils.type_converters import build_decimal_converter, build_uuid_coercions, time_iso_convert from sqlspec.utils.type_guards import has_rowcount +from sqlspec.utils.uuids import uuid_from_string if TYPE_CHECKING: from collections.abc import Callable, Mapping @@ -395,7 +395,7 @@ def _restore_uuid_columns(rows: "list[dict[str, Any]]", description: "list[Any] for column in uuid_columns: value = row.get(column) if isinstance(value, str): - row[column] = UUID(value) + row[column] = uuid_from_string(value) driver_profile = build_profile() diff --git a/sqlspec/adapters/oracledb/_uuid_handlers.py b/sqlspec/adapters/oracledb/_uuid_handlers.py index 47459e78e..a0434f4ce 100644 --- a/sqlspec/adapters/oracledb/_uuid_handlers.py +++ b/sqlspec/adapters/oracledb/_uuid_handlers.py @@ -10,6 +10,7 @@ from sqlspec.adapters.oracledb._json_handlers import chain_input_handler, chain_output_handler from sqlspec.adapters.oracledb._typing import DB_TYPE_RAW from sqlspec.utils.logging import get_logger +from sqlspec.utils.uuids import uuid_from_bytes if TYPE_CHECKING: from oracledb import AsyncConnection, AsyncCursor, Connection, Cursor @@ -59,7 +60,7 @@ def uuid_converter_out(value: bytes | None) -> "uuid.UUID | bytes | None": return value try: - return uuid.UUID(bytes=value) + return uuid_from_bytes(value) except (ValueError, TypeError): logger.debug("RAW(16) value is not valid UUID format, returning as bytes", extra={"value_length": len(value)}) return value diff --git a/sqlspec/adapters/psqlpy/core.py b/sqlspec/adapters/psqlpy/core.py index 2bd7bd960..498adf9db 100644 --- a/sqlspec/adapters/psqlpy/core.py +++ b/sqlspec/adapters/psqlpy/core.py @@ -50,6 +50,7 @@ from sqlspec.utils.text import quote_identifier, split_qualified_identifier from sqlspec.utils.type_converters import build_nested_decimal_normalizer, build_uuid_coercions from sqlspec.utils.type_guards import has_query_result_metadata +from sqlspec.utils.uuids import uuid_from_string if TYPE_CHECKING: from collections.abc import Callable, Mapping @@ -559,7 +560,7 @@ def _coerce_uuid_parameter(value: Any) -> Any: return value if isinstance(value, str): try: - return uuid.UUID(value) + return uuid_from_string(value) except ValueError as error: msg = "Invalid UUID parameter for psqlpy." raise SQLSpecError(msg) from error diff --git a/sqlspec/adapters/spanner/type_converter.py b/sqlspec/adapters/spanner/type_converter.py index 38c0fce81..8bdcf4f00 100644 --- a/sqlspec/adapters/spanner/type_converter.py +++ b/sqlspec/adapters/spanner/type_converter.py @@ -24,6 +24,7 @@ from sqlspec.core import TypedParameter from sqlspec.utils.module_loader import import_optional_attr from sqlspec.utils.type_converters import should_json_encode_sequence +from sqlspec.utils.uuids import uuid_from_bytes if TYPE_CHECKING: from collections.abc import Callable @@ -146,7 +147,7 @@ def spanner_to_uuid(value: "bytes | None") -> "UUID | bytes | None": if len(value) != UUID_BYTE_LENGTH: return value try: - return UUID(bytes=value) + return uuid_from_bytes(value) except (ValueError, TypeError): return value @@ -194,7 +195,7 @@ def coerce_params_for_spanner( if type(value) is TypedParameter: value = value.value if isinstance(value, _UUID_TYPES): - std_uuid = value if isinstance(value, UUID) else UUID(bytes=value.bytes) + std_uuid = value if isinstance(value, UUID) else uuid_from_bytes(value.bytes) coerced[key] = bytes_to_spanner(uuid_to_spanner(std_uuid)) elif isinstance(value, bytes): coerced[key] = bytes_to_spanner(value) diff --git a/sqlspec/core/type_converter.py b/sqlspec/core/type_converter.py index 1c69d4db8..e340e9334 100644 --- a/sqlspec/core/type_converter.py +++ b/sqlspec/core/type_converter.py @@ -8,6 +8,8 @@ from mypy_extensions import mypyc_attr +from sqlspec.utils.uuids import uuid_from_string + __all__ = ( "BaseInputConverter", "convert_decimal", @@ -30,7 +32,7 @@ def convert_uuid(value: str) -> UUID: Returns: UUID object. """ - return UUID(value) + return uuid_from_string(value) def convert_iso_datetime(value: str) -> "datetime": diff --git a/sqlspec/utils/arrow_helpers.py b/sqlspec/utils/arrow_helpers.py index a82235e5f..ea537d30f 100644 --- a/sqlspec/utils/arrow_helpers.py +++ b/sqlspec/utils/arrow_helpers.py @@ -11,12 +11,12 @@ from collections.abc import Callable, Iterable, Mapping from functools import lru_cache from typing import TYPE_CHECKING, Any, Literal, cast, overload -from uuid import UUID from sqlspec.exceptions import ImproperConfigurationError from sqlspec.utils.dispatch import TypeDispatcher from sqlspec.utils.module_loader import ensure_pandas, ensure_polars, ensure_pyarrow from sqlspec.utils.type_guards import has_arrow_table_stats, has_get_data +from sqlspec.utils.uuids import uuid_from_bytes if TYPE_CHECKING: from sqlspec.core.result import ArrowResult @@ -546,11 +546,11 @@ def _arrow_uuid_column_to_pylist(column: Any, data_type: Any) -> "list[Any]": array = cast("Any", column.combine_chunks() if isinstance(column, pa.ChunkedArray) else column) storage_values = cast("list[Any]", array.storage.to_pylist()) - return [UUID(bytes=value) if value is not None else None for value in storage_values] + return [uuid_from_bytes(value) if value is not None else None for value in storage_values] if _arrow_type_is_list_of_opaque_uuid(data_type): nested_values = cast("list[Any]", column.to_pylist()) return [ - [UUID(bytes=item) if item is not None else None for item in value] if value is not None else None + [uuid_from_bytes(item) if item is not None else None for item in value] if value is not None else None for value in nested_values ] return cast("list[Any]", column.to_pylist()) diff --git a/sqlspec/utils/schema.py b/sqlspec/utils/schema.py index 2539893f7..8cbde3ef3 100644 --- a/sqlspec/utils/schema.py +++ b/sqlspec/utils/schema.py @@ -29,6 +29,7 @@ is_pydantic_model, is_typed_dict, ) +from sqlspec.utils.uuids import uuid_from_bytes, uuid_from_string __all__ = ( "_DEFAULT_TYPE_DECODERS", @@ -733,12 +734,12 @@ def _convert_to_uuid(value: Any) -> UUID: return value if isinstance(value, str): try: - return UUID(value) + return uuid_from_string(value) except ValueError: pass if isinstance(value, bytes): try: - return UUID(bytes=value) + return uuid_from_bytes(value) except ValueError: pass msg = f"Cannot convert {type(value).__name__} to UUID" diff --git a/sqlspec/utils/type_converters.py b/sqlspec/utils/type_converters.py index a4cc25340..73e45a6be 100644 --- a/sqlspec/utils/type_converters.py +++ b/sqlspec/utils/type_converters.py @@ -8,6 +8,7 @@ from sqlspec.utils.dispatch import TypeDispatcher from sqlspec.utils.module_loader import import_optional_attr +from sqlspec.utils.uuids import uuid_from_int if TYPE_CHECKING: import datetime @@ -189,7 +190,7 @@ def _uuid_to_string(value: object) -> str: def _uuid_utils_to_stdlib(value: _UUIDLike) -> UUID: - return UUID(int=value.int) + return uuid_from_int(value.int) def build_uuid_coercions(*, native: bool = False) -> "dict[type[Any], Callable[[Any], Any]]": diff --git a/sqlspec/utils/uuids.py b/sqlspec/utils/uuids.py index d7da93cd7..462aeb904 100644 --- a/sqlspec/utils/uuids.py +++ b/sqlspec/utils/uuids.py @@ -1,8 +1,8 @@ """UUID and ID generation utilities with optional acceleration. -Provides wrapper functions for uuid3, uuid4, uuid5, uuid6, uuid7, and nanoid generation. -Uses uuid-utils and fastnanoid packages for performance when available, -falling back to standard library. +Provides wrapper functions for UUID construction, uuid3, uuid4, uuid5, uuid6, +uuid7, and nanoid generation. Uses uuid-utils and fastnanoid packages for +performance when available, falling back to the standard library. When uuid-utils is installed: - uuid3, uuid4, uuid5, uuid6, uuid7 use the faster Rust implementation @@ -43,13 +43,60 @@ "uuid5", "uuid6", "uuid7", + "uuid_from_bytes", + "uuid_from_int", + "uuid_from_string", ) _uuid_utils_mod: Any | None = import_optional("uuid_utils.compat") +_uuid_utils_native_mod: Any | None = import_optional("uuid_utils") _fastnanoid_mod: Any | None = import_optional("fastnanoid") +def uuid_from_string(value: str) -> "UUID": + """Construct a stdlib UUID from text, using Rust parsing when available. + + Args: + value: Canonical, hexadecimal, URN, or braced UUID text accepted by + ``uuid.UUID`` and ``uuid_utils.UUID``. + + Returns: + A standard-library UUID suitable for native database drivers. + """ + module = _uuid_utils_native_mod + if module is None: + return UUID(value) + return UUID(int=module.UUID(value).int) + + +def uuid_from_bytes(value: bytes) -> "UUID": + """Construct a stdlib UUID from its 16-byte representation. + + The stdlib constructor is retained for this shape because converting a + Rust UUID back to the driver-compatible stdlib type is slower. + + Args: + value: UUID bytes in big-endian order. + + Returns: + A standard-library UUID. + """ + return UUID(bytes=value) + + +def uuid_from_int(value: int) -> "UUID": + """Construct a stdlib UUID from its 128-bit integer value. + + Args: + value: UUID integer in the inclusive range ``0`` through ``2**128 - 1``. + + Returns: + A standard-library UUID. + """ + return UUID(int=value) + + def uuid3(name: str, namespace: "UUID | None" = None) -> "UUID": """Generate a deterministic UUID (version 3) using MD5 hash. diff --git a/tests/unit/adapters/test_oracledb/test_uuid_handlers.py b/tests/unit/adapters/test_oracledb/test_uuid_handlers.py index 588bd3348..56fa5497b 100644 --- a/tests/unit/adapters/test_oracledb/test_uuid_handlers.py +++ b/tests/unit/adapters/test_oracledb/test_uuid_handlers.py @@ -53,7 +53,7 @@ def test_uuid_converter_out_type_error() -> None: """TypeError should fall back to original bytes.""" payload = b"1234567890123456" - with patch("uuid.UUID", side_effect=TypeError("Invalid type")): + with patch("sqlspec.adapters.oracledb._uuid_handlers.uuid_from_bytes", side_effect=TypeError("Invalid type")): result = uuid_converter_out(payload) assert result is payload @@ -62,7 +62,7 @@ def test_uuid_converter_out_value_error() -> None: """ValueError should fall back to original bytes.""" payload = b"1234567890123456" - with patch("uuid.UUID", side_effect=ValueError("Invalid UUID")): + with patch("sqlspec.adapters.oracledb._uuid_handlers.uuid_from_bytes", side_effect=ValueError("Invalid UUID")): result = uuid_converter_out(payload) assert result is payload diff --git a/tests/unit/utils/test_uuids.py b/tests/unit/utils/test_uuids.py index 28344115c..b36ed1587 100644 --- a/tests/unit/utils/test_uuids.py +++ b/tests/unit/utils/test_uuids.py @@ -4,7 +4,9 @@ via uuid-utils and fastnanoid packages. """ +import ast import warnings +from pathlib import Path from uuid import UUID import pytest @@ -23,6 +25,9 @@ uuid5, uuid6, uuid7, + uuid_from_bytes, + uuid_from_int, + uuid_from_string, ) @@ -68,6 +73,77 @@ def test_uuid4_returns_valid_uuid() -> None: assert _is_uuid_like(result) +def test_uuid_construction_helpers_return_stdlib_uuid() -> None: + """Central constructors preserve the public and driver-compatible UUID type.""" + expected = UUID("12345678-1234-5678-1234-567812345678") + + assert uuid_from_string(str(expected)) == expected + assert uuid_from_bytes(expected.bytes) == expected + assert uuid_from_int(expected.int) == expected + assert type(uuid_from_string(str(expected))) is UUID + assert type(uuid_from_bytes(expected.bytes)) is UUID + assert type(uuid_from_int(expected.int)) is UUID + + +def test_uuid_from_string_uses_cached_rust_parser(monkeypatch: pytest.MonkeyPatch) -> None: + """String parsing uses uuid-utils internally while returning stdlib UUID.""" + import sqlspec.utils.uuids as _uuids + + expected = UUID("12345678-1234-5678-1234-567812345678") + calls: list[str] = [] + + class RustUuid: + def __init__(self, value: str) -> None: + calls.append(value) + self.int = expected.int + + class FakeUuidUtils: + UUID = RustUuid + + monkeypatch.setattr(_uuids, "_uuid_utils_native_mod", FakeUuidUtils()) + + result = _uuids.uuid_from_string(str(expected)) + + assert result == expected + assert type(result) is UUID + assert calls == [str(expected)] + + +def test_production_uuid_construction_is_centralized() -> None: + """Production modules construct and generate UUIDs only through the facade.""" + source_root = Path(__file__).parents[3] / "sqlspec" + violations: list[str] = [] + for path in source_root.rglob("*.py"): + if path == source_root / "utils" / "uuids.py": + continue + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module == "uuid": + if any(alias.name == "uuid4" for alias in node.names): + violations.append(f"{path.relative_to(source_root)}:{node.lineno}") + continue + if not isinstance(node, ast.Call): + continue + function = node.func + is_direct = isinstance(function, ast.Name) and function.id == "UUID" + is_module = ( + isinstance(function, ast.Attribute) + and function.attr == "UUID" + and isinstance(function.value, ast.Name) + and function.value.id in {"uuid", "uuid_mod", "_uuid_mod"} + ) + is_module_uuid4 = ( + isinstance(function, ast.Attribute) + and function.attr == "uuid4" + and isinstance(function.value, ast.Name) + and function.value.id in {"uuid", "uuid_mod", "_uuid_mod"} + ) + if is_direct or is_module or is_module_uuid4: + violations.append(f"{path.relative_to(source_root)}:{node.lineno}") + + assert violations == [] + + def test_accelerated_uuid_helpers_return_stdlib_uuid_instances() -> None: """The public return annotation is truthful when uuid-utils accelerates generation.""" results = (uuid3("test-name"), uuid4(), uuid5("test-name"), uuid6(), uuid7()) From 95c1030ba5d1e9402711c2cc0d27bf6c81191b8f Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 19:48:54 +0000 Subject: [PATCH 09/15] perf(oracledb): avoid pass-through bind copies --- sqlspec/adapters/oracledb/core.py | 370 ++++++++++++++---- .../test_oracledb/test_lob_coercion.py | 81 +++- 2 files changed, 367 insertions(+), 84 deletions(-) diff --git a/sqlspec/adapters/oracledb/core.py b/sqlspec/adapters/oracledb/core.py index ca8120a2b..c2aa9bb28 100644 --- a/sqlspec/adapters/oracledb/core.py +++ b/sqlspec/adapters/oracledb/core.py @@ -2,7 +2,7 @@ import contextlib import re -from collections.abc import Sized +from collections.abc import Iterable, Sized from typing import TYPE_CHECKING, Any, Protocol, cast from sqlspec.adapters.oracledb._json_handlers import is_json_payload @@ -93,6 +93,7 @@ TYPE_CONVERTER = OracleOutputConverter() _LOB_TYPE_NAME_MARKERS: "tuple[str, ...]" = ("LOB", "BFILE") _SCALAR_PASSTHROUGH_TYPES: "tuple[type[Any], ...]" = (bool, int, float, str, bytes, bytearray, type(None)) +_BIND_PASSTHROUGH_TYPES: "tuple[type[Any], ...]" = (bool, int, float, type(None)) ROW_CACHE_MAX_SIZE: int = 256 # Oracle ORA error code ranges for category detection @@ -255,6 +256,30 @@ def normalize_execute_many_parameters_async(parameters: Any) -> Any: return parameters +class _OracleJsonBindingState: + """Lazily resolve whether JSON-shaped values require BLOB locators.""" + + __slots__ = ("_connection", "_resolved", "_use_blob", "_version_cache") + + def __init__(self, connection: Any, version_cache: Any) -> None: + self._connection = connection + self._version_cache = version_cache + self._resolved = False + self._use_blob = False + + def uses_blob(self) -> bool: + """Return the operation-scoped JSON storage decision.""" + if not self._resolved: + server_major = resolve_oracle_connection_major(self._connection, self._version_cache) + self._use_blob = ( + isinstance(server_major, int) + and not isinstance(server_major, bool) + and resolve_oracle_json_storage(server_major) == ORACLE_JSON_STORAGE_BLOB_JSON + ) + self._resolved = True + return self._use_blob + + def coerce_large_parameters_sync( connection: Any, parameters: Any, @@ -291,32 +316,15 @@ def coerce_large_parameters_sync( """ if not parameters: return parameters - if isinstance(parameters, dict): - for param_name, param_value in parameters.items(): - parameters[param_name] = _coerce_value_sync( - connection, - param_value, - clob_type=clob_type, - blob_type=blob_type, - varchar2_byte_limit=varchar2_byte_limit, - raw_byte_limit=raw_byte_limit, - version_cache=version_cache, - ) - return parameters - if isinstance(parameters, (list, tuple)): - return [ - _coerce_value_sync( - connection, - value, - clob_type=clob_type, - blob_type=blob_type, - varchar2_byte_limit=varchar2_byte_limit, - raw_byte_limit=raw_byte_limit, - version_cache=version_cache, - ) - for value in parameters - ] - return parameters + return _coerce_parameters_sync( + connection, + parameters, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=_OracleJsonBindingState(connection, version_cache), + ) async def coerce_large_parameters_async( @@ -336,32 +344,15 @@ async def coerce_large_parameters_async( """ if not parameters: return parameters - if isinstance(parameters, dict): - for param_name, param_value in parameters.items(): - parameters[param_name] = await _coerce_value_async( - connection, - param_value, - clob_type=clob_type, - blob_type=blob_type, - varchar2_byte_limit=varchar2_byte_limit, - raw_byte_limit=raw_byte_limit, - version_cache=version_cache, - ) - return parameters - if isinstance(parameters, (list, tuple)): - return [ - await _coerce_value_async( - connection, - value, - clob_type=clob_type, - blob_type=blob_type, - varchar2_byte_limit=varchar2_byte_limit, - raw_byte_limit=raw_byte_limit, - version_cache=version_cache, - ) - for value in parameters - ] - return parameters + return await _coerce_parameters_async( + connection, + parameters, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=_OracleJsonBindingState(connection, version_cache), + ) def coerce_many_parameters_sync( @@ -378,18 +369,24 @@ def coerce_many_parameters_sync( normalized = normalize_execute_many_parameters_sync(parameters) if not normalized: return normalized - return [ - coerce_large_parameters_sync( + json_binding_state = _OracleJsonBindingState(connection, version_cache) + coerced_rows: list[Any] | None = None + for index, row in enumerate(normalized): + coerced_row = _coerce_parameters_sync( connection, row, clob_type=clob_type, blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, - version_cache=version_cache, + json_binding_state=json_binding_state, ) - for row in normalized - ] + if coerced_rows is None: + if coerced_row is row: + continue + coerced_rows = list(normalized[:index]) + coerced_rows.append(coerced_row) + return normalized if coerced_rows is None else coerced_rows async def coerce_many_parameters_async( @@ -406,18 +403,24 @@ async def coerce_many_parameters_async( normalized = normalize_execute_many_parameters_async(parameters) if not normalized: return normalized - return [ - await coerce_large_parameters_async( + json_binding_state = _OracleJsonBindingState(connection, version_cache) + coerced_rows: list[Any] | None = None + for index, row in enumerate(normalized): + coerced_row = await _coerce_parameters_async( connection, row, clob_type=clob_type, blob_type=blob_type, varchar2_byte_limit=varchar2_byte_limit, raw_byte_limit=raw_byte_limit, - version_cache=version_cache, + json_binding_state=json_binding_state, ) - for row in normalized - ] + if coerced_rows is None: + if coerced_row is row: + continue + coerced_rows = list(normalized[:index]) + coerced_rows.append(coerced_row) + return normalized if coerced_rows is None else coerced_rows def build_insert_statement(table: str, columns: "list[str]") -> str: @@ -952,6 +955,165 @@ def _check_pending_exception(self, exc_handler: "AsyncExceptionHandler") -> None def _resolve_row_metadata(self, description: object) -> tuple[list[str], bool]: ... +def _parameter_values_need_coercion( + values: "Iterable[Any]", + *, + varchar2_byte_limit: int, + raw_byte_limit: int, + json_binding_state: _OracleJsonBindingState, +) -> bool: + """Return whether any value requires wrapper, LOB, or JSON rewriting.""" + for value in values: + value_type = type(value) + if value_type in _BIND_PASSTHROUGH_TYPES: + continue + if value_type is str: + if len(value.encode("utf-8")) > varchar2_byte_limit: + return True + continue + if value_type is bytes or value_type is bytearray: + if len(value) > raw_byte_limit: + return True + continue + if value_type is dict or value_type is list or value_type is tuple: + if is_json_payload(value) and json_binding_state.uses_blob(): + return True + continue + if isinstance(value, (OracleClob, OracleBlob, OracleJson)): + return True + if isinstance(value, str): + if len(value.encode("utf-8")) > varchar2_byte_limit: + return True + continue + if isinstance(value, (bytes, bytearray)): + if len(value) > raw_byte_limit: + return True + continue + if is_json_payload(value) and json_binding_state.uses_blob(): + return True + return False + + +def _coerce_parameters_sync( + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + json_binding_state: _OracleJsonBindingState, +) -> Any: + """Coerce one parameter container, copying sequences only when changed.""" + if isinstance(parameters, dict): + if not _parameter_values_need_coercion( + parameters.values(), + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ): + return parameters + for param_name, param_value in parameters.items(): + coerced_value = _coerce_value_sync( + connection, + param_value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) + if coerced_value is not param_value: + parameters[param_name] = coerced_value + return parameters + if isinstance(parameters, (list, tuple)): + if not _parameter_values_need_coercion( + parameters, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ): + return parameters + coerced_values: list[Any] | None = None + for index, value in enumerate(parameters): + coerced_value = _coerce_value_sync( + connection, + value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) + if coerced_values is None: + if coerced_value is value: + continue + coerced_values = list(parameters[:index]) + coerced_values.append(coerced_value) + return parameters if coerced_values is None else coerced_values + return parameters + + +async def _coerce_parameters_async( + connection: Any, + parameters: Any, + *, + clob_type: Any, + blob_type: Any, + varchar2_byte_limit: int, + raw_byte_limit: int, + json_binding_state: _OracleJsonBindingState, +) -> Any: + """Async mirror of :func:`_coerce_parameters_sync`.""" + if isinstance(parameters, dict): + if not _parameter_values_need_coercion( + parameters.values(), + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ): + return parameters + for param_name, param_value in parameters.items(): + coerced_value = await _coerce_value_async( + connection, + param_value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) + if coerced_value is not param_value: + parameters[param_name] = coerced_value + return parameters + if isinstance(parameters, (list, tuple)): + if not _parameter_values_need_coercion( + parameters, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ): + return parameters + coerced_values: list[Any] | None = None + for index, value in enumerate(parameters): + coerced_value = await _coerce_value_async( + connection, + value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) + if coerced_values is None: + if coerced_value is value: + continue + coerced_values = list(parameters[:index]) + coerced_values.append(coerced_value) + return parameters if coerced_values is None else coerced_values + return parameters + + def _coerce_value_sync( connection: Any, value: Any, @@ -960,9 +1122,28 @@ def _coerce_value_sync( blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int, - version_cache: Any = None, + json_binding_state: _OracleJsonBindingState, ) -> Any: """Route a single parameter value through wrapper-aware coercion (sync).""" + value_type = type(value) + if value_type in _BIND_PASSTHROUGH_TYPES: + return value + if value_type is str: + if len(value.encode("utf-8")) > varchar2_byte_limit: + return connection.createlob(clob_type, value) + return value + if value_type is bytes: + if len(value) > raw_byte_limit: + return connection.createlob(blob_type, value) + return value + if value_type is bytearray: + if len(value) > raw_byte_limit: + return connection.createlob(blob_type, bytes(value)) + return value + if value_type is dict or value_type is list or value_type is tuple: + if is_json_payload(value) and json_binding_state.uses_blob(): + return connection.createlob(blob_type, to_json(value, as_bytes=True)) + return value if isinstance(value, OracleClob): inner = value.value if isinstance(inner, bytes): @@ -974,19 +1155,21 @@ def _coerce_value_sync( inner = inner.encode("utf-8") return connection.createlob(blob_type, inner) if isinstance(value, OracleJson): - value = value.value - server_major = resolve_oracle_connection_major(connection, version_cache) - if ( - isinstance(server_major, int) - and not isinstance(server_major, bool) - and resolve_oracle_json_storage(server_major) == ORACLE_JSON_STORAGE_BLOB_JSON - and is_json_payload(value) - ): - return connection.createlob(blob_type, to_json(value, as_bytes=True)) + return _coerce_value_sync( + connection, + value.value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) if isinstance(value, str) and len(value.encode("utf-8")) > varchar2_byte_limit: return connection.createlob(clob_type, value) if isinstance(value, (bytes, bytearray)) and len(value) > raw_byte_limit: return connection.createlob(blob_type, bytes(value)) + if is_json_payload(value) and json_binding_state.uses_blob(): + return connection.createlob(blob_type, to_json(value, as_bytes=True)) return value @@ -998,9 +1181,28 @@ async def _coerce_value_async( blob_type: Any, varchar2_byte_limit: int, raw_byte_limit: int, - version_cache: Any = None, + json_binding_state: _OracleJsonBindingState, ) -> Any: """Async mirror of :func:`_coerce_value_sync`.""" + value_type = type(value) + if value_type in _BIND_PASSTHROUGH_TYPES: + return value + if value_type is str: + if len(value.encode("utf-8")) > varchar2_byte_limit: + return await connection.createlob(clob_type, value) + return value + if value_type is bytes: + if len(value) > raw_byte_limit: + return await connection.createlob(blob_type, value) + return value + if value_type is bytearray: + if len(value) > raw_byte_limit: + return await connection.createlob(blob_type, bytes(value)) + return value + if value_type is dict or value_type is list or value_type is tuple: + if is_json_payload(value) and json_binding_state.uses_blob(): + return await connection.createlob(blob_type, to_json(value, as_bytes=True)) + return value if isinstance(value, OracleClob): inner = value.value if isinstance(inner, bytes): @@ -1012,19 +1214,21 @@ async def _coerce_value_async( inner = inner.encode("utf-8") return await connection.createlob(blob_type, inner) if isinstance(value, OracleJson): - value = value.value - server_major = resolve_oracle_connection_major(connection, version_cache) - if ( - isinstance(server_major, int) - and not isinstance(server_major, bool) - and resolve_oracle_json_storage(server_major) == ORACLE_JSON_STORAGE_BLOB_JSON - and is_json_payload(value) - ): - return await connection.createlob(blob_type, to_json(value, as_bytes=True)) + return await _coerce_value_async( + connection, + value.value, + clob_type=clob_type, + blob_type=blob_type, + varchar2_byte_limit=varchar2_byte_limit, + raw_byte_limit=raw_byte_limit, + json_binding_state=json_binding_state, + ) if isinstance(value, str) and len(value.encode("utf-8")) > varchar2_byte_limit: return await connection.createlob(clob_type, value) if isinstance(value, (bytes, bytearray)) and len(value) > raw_byte_limit: return await connection.createlob(blob_type, bytes(value)) + if is_json_payload(value) and json_binding_state.uses_blob(): + return await connection.createlob(blob_type, to_json(value, as_bytes=True)) return value diff --git a/tests/unit/adapters/test_oracledb/test_lob_coercion.py b/tests/unit/adapters/test_oracledb/test_lob_coercion.py index 92c708410..32bd33967 100644 --- a/tests/unit/adapters/test_oracledb/test_lob_coercion.py +++ b/tests/unit/adapters/test_oracledb/test_lob_coercion.py @@ -2,7 +2,7 @@ from collections.abc import Callable from typing import cast -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -198,6 +198,44 @@ def test_coerce_many_parameters_sync_applies_json_coercion_to_every_row(sync_con assert all(row["payload"] is sync_connection.createlob.return_value for row in result) +def test_coerce_many_parameters_sync_preserves_unchanged_rows_without_version_resolution( + sync_connection: MagicMock, +) -> None: + """Pass-through batches retain their containers and skip JSON metadata work.""" + rows = [(1, "short", b"x", None, 3.5), (2, "short", b"x", None, 4.5)] + + with patch("sqlspec.adapters.oracledb.core.resolve_oracle_connection_major") as resolve_major: + result = coerce_many_parameters_sync( + sync_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + assert result is rows + assert all(result_row is source_row for result_row, source_row in zip(result, rows, strict=True)) + resolve_major.assert_not_called() + + +def test_coerce_many_parameters_sync_resolves_json_storage_once(sync_connection: MagicMock) -> None: + """One batch shares a single server-storage decision across all JSON values.""" + rows = [{"id": 1, "payload": {"row": 1}}, {"id": 2, "payload": [{"row": 2}]}] + + with patch("sqlspec.adapters.oracledb.core.resolve_oracle_connection_major", return_value=18) as resolve_major: + coerce_many_parameters_sync( + sync_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + resolve_major.assert_called_once_with(sync_connection, None) + + @pytest.mark.anyio async def test_coerce_many_parameters_async_applies_json_coercion_to_every_row(async_connection: AsyncMock) -> None: """Async executemany coercion awaits locator creation for every row.""" @@ -217,6 +255,46 @@ async def test_coerce_many_parameters_async_applies_json_coercion_to_every_row(a assert all(row["payload"] is async_connection.createlob.return_value for row in result) +@pytest.mark.anyio +async def test_coerce_many_parameters_async_preserves_unchanged_rows_without_version_resolution( + async_connection: AsyncMock, +) -> None: + """Async pass-through batches retain containers and skip JSON metadata work.""" + rows = [(1, "short", b"x", None, 3.5), (2, "short", b"x", None, 4.5)] + + with patch("sqlspec.adapters.oracledb.core.resolve_oracle_connection_major") as resolve_major: + result = await coerce_many_parameters_async( + async_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + assert result is rows + assert all(result_row is source_row for result_row, source_row in zip(result, rows, strict=True)) + resolve_major.assert_not_called() + + +@pytest.mark.anyio +async def test_coerce_many_parameters_async_resolves_json_storage_once(async_connection: AsyncMock) -> None: + """Async batches share one server-storage decision across JSON values.""" + rows = [{"id": 1, "payload": {"row": 1}}, {"id": 2, "payload": [{"row": 2}]}] + + with patch("sqlspec.adapters.oracledb.core.resolve_oracle_connection_major", return_value=18) as resolve_major: + await coerce_many_parameters_async( + async_connection, + rows, + clob_type=CLOB_TYPE, + blob_type=BLOB_TYPE, + varchar2_byte_limit=VARCHAR2_LIMIT, + raw_byte_limit=RAW_LIMIT, + ) + + resolve_major.assert_called_once_with(async_connection, None) + + def test_oracle_sync_stream_source_coerces_json_filter_before_execute() -> None: """Native sync streaming uses the same Oracle 18c BLOB locator bind path.""" locator = MagicMock(name="BLOB") @@ -277,6 +355,7 @@ def test_coerce_large_parameters_sync_list_parameters_passthrough(sync_connectio raw_byte_limit=RAW_LIMIT, ) assert result == ["a", "b"] + assert result is params sync_connection.createlob.assert_not_called() From dbf3bcd9f85e9c22624f65560dedd4135d3b4aeb Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 20:26:12 +0000 Subject: [PATCH 10/15] perf(bigquery): preserve validated record batches --- sqlspec/adapters/bigquery/driver.py | 22 +++++- .../test_bigquery/test_job_controls.py | 68 ++++++++++++++++++- 2 files changed, 87 insertions(+), 3 deletions(-) diff --git a/sqlspec/adapters/bigquery/driver.py b/sqlspec/adapters/bigquery/driver.py index f90b7997e..fd9edb761 100644 --- a/sqlspec/adapters/bigquery/driver.py +++ b/sqlspec/adapters/bigquery/driver.py @@ -772,14 +772,32 @@ def _bigquery_arrow_reader_from_iterable(batches: "Iterable[ArrowRecordBatch]") def _records_to_json_rows( records: "Sequence[Mapping[str, Any]] | Sequence[Sequence[Any]]", columns: "list[str] | None" ) -> "list[dict[str, Any]]": - materialized = list(records) + materialized = records if isinstance(records, list) else list(records) if not materialized: msg = "load_from_records requires at least one record." raise ImproperConfigurationError(msg) first = materialized[0] if isinstance(first, Mapping): - resolved = columns if columns is not None else list(first.keys()) + if columns is None: + resolved = list(first.keys()) + expected = set(resolved) + can_reuse = True + for record in materialized: + if not isinstance(record, Mapping): + msg = "load_from_records mapping records must all be mappings." + raise ImproperConfigurationError(msg) + record_keys = list(record.keys()) + if set(record_keys) != expected: + msg = "load_from_records mapping records must all share the same keys." + raise ImproperConfigurationError(msg) + if type(record) is not dict or record_keys != resolved: + can_reuse = False + if can_reuse: + return cast("list[dict[str, Any]]", materialized) + return [{column: record[column] for column in resolved} for record in materialized] + + resolved = columns expected = set(resolved) rows: list[dict[str, Any]] = [] for record in materialized: diff --git a/tests/unit/adapters/test_bigquery/test_job_controls.py b/tests/unit/adapters/test_bigquery/test_job_controls.py index 66d4f11f0..272d2c776 100644 --- a/tests/unit/adapters/test_bigquery/test_job_controls.py +++ b/tests/unit/adapters/test_bigquery/test_job_controls.py @@ -1,14 +1,17 @@ """Unit tests for BigQuery job-control behavior.""" +from collections import UserDict from types import SimpleNamespace from typing import Any, cast import pyarrow as pa +import pytest from google.cloud.bigquery import LoadJobConfig from google.cloud.bigquery.enums import QueryApiMethod, TimestampPrecision from sqlspec.adapters.bigquery.core import build_load_job_config, run_query_job, try_bulk_insert -from sqlspec.adapters.bigquery.driver import BigQueryDriver +from sqlspec.adapters.bigquery.driver import BigQueryDriver, _records_to_json_rows +from sqlspec.exceptions import ImproperConfigurationError from sqlspec.utils.serializers import to_json CAPABILITIES = { @@ -250,6 +253,69 @@ def test_load_from_records_uses_bigquery_json_load_api() -> None: assert connection.load_job.result_calls[0]["timeout"] == driver._job_request_timeout() +def test_records_to_json_rows_preserves_validated_plain_dict_batch() -> None: + records = [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}] + + rows = _records_to_json_rows(records, None) + + assert rows is records + assert rows[0] is records[0] + assert rows[1] is records[1] + + +def test_records_to_json_rows_rebuilds_rows_with_different_key_order() -> None: + records = [{"id": 1, "name": "a"}, {"name": "b", "id": 2}] + + rows = _records_to_json_rows(records, None) + + assert rows is not records + assert rows == [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}] + assert list(rows[1]) == ["id", "name"] + + +def test_records_to_json_rows_rebuilds_rows_for_explicit_columns() -> None: + records = [{"id": 1, "name": "a"}] + + rows = _records_to_json_rows(records, ["name", "id"]) + + assert rows is not records + assert rows == [{"name": "a", "id": 1}] + assert list(rows[0]) == ["name", "id"] + + +def test_records_to_json_rows_rebuilds_mapping_subclasses() -> None: + record = UserDict({"id": 1, "name": "a"}) + + rows = _records_to_json_rows([record], None) + + assert rows == [{"id": 1, "name": "a"}] + assert rows[0] is not record + assert type(rows[0]) is dict + + +def test_records_to_json_rows_materializes_non_list_input_once() -> None: + records = ({"id": value} for value in range(2)) + + rows = _records_to_json_rows(cast(Any, records), None) + + assert rows == [{"id": 0}, {"id": 1}] + + +@pytest.mark.parametrize( + ("records", "columns", "message"), + [ + ([], None, "at least one record"), + ([{"id": 1}, (2,)], None, "must all be mappings"), + ([{"id": 1}, {"name": "a"}], None, "must all share the same keys"), + ([(1, "a")], None, "requires columns"), + ([(1, "a")], ["id"], "must match the number of columns"), + ], +) +def test_records_to_json_rows_preserves_validation_errors(records: Any, columns: Any, message: str) -> None: + with pytest.raises(ImproperConfigurationError, match=message): + _records_to_json_rows(records, columns) + + def test_load_job_config_fill_from_default_preserves_defaults() -> None: default_job_config = LoadJobConfig(labels={"source": "default"}) From a48dc1a2fddf39c6dd33201ecc8986275409aa01 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 20:40:07 +0000 Subject: [PATCH 11/15] perf(spanner): preserve driver-ready parameters --- sqlspec/adapters/spanner/type_converter.py | 14 +++- .../test_spanner/test_type_converter.py | 80 ++++++++++++++++++- 2 files changed, 91 insertions(+), 3 deletions(-) diff --git a/sqlspec/adapters/spanner/type_converter.py b/sqlspec/adapters/spanner/type_converter.py index 8bdcf4f00..478726261 100644 --- a/sqlspec/adapters/spanner/type_converter.py +++ b/sqlspec/adapters/spanner/type_converter.py @@ -191,28 +191,38 @@ def coerce_params_for_spanner( json_object_type = _get_json_object_type() coerced: dict[str, Any] = {} + changed = False for key, value in params.items(): if type(value) is TypedParameter: value = value.value + changed = True if isinstance(value, _UUID_TYPES): std_uuid = value if isinstance(value, UUID) else uuid_from_bytes(value.bytes) coerced[key] = bytes_to_spanner(uuid_to_spanner(std_uuid)) + changed = True elif isinstance(value, bytes): coerced[key] = bytes_to_spanner(value) + changed = True elif isinstance(value, datetime) and value.tzinfo is None: coerced[key] = value.replace(tzinfo=timezone.utc) + changed = True elif isinstance(value, json_object_type): coerced[key] = value elif isinstance(value, dict): coerced[key] = spanner_json(value) + changed = True elif isinstance(value, (list, tuple)): if should_json_encode_sequence(value): coerced[key] = spanner_json(list(value)) + changed = True + elif isinstance(value, tuple): + coerced[key] = list(value) + changed = True else: - coerced[key] = list(value) if isinstance(value, tuple) else value + coerced[key] = value else: coerced[key] = value - return coerced + return coerced if changed else params def infer_spanner_param_types(params: "dict[str, Any] | None") -> "dict[str, Any]": diff --git a/tests/unit/adapters/test_spanner/test_type_converter.py b/tests/unit/adapters/test_spanner/test_type_converter.py index fed7e791a..662b4e6aa 100644 --- a/tests/unit/adapters/test_spanner/test_type_converter.py +++ b/tests/unit/adapters/test_spanner/test_type_converter.py @@ -1,6 +1,10 @@ -from datetime import datetime, timezone +import base64 +from datetime import date, datetime, timezone from uuid import UUID +import uuid_utils +from google.cloud.spanner_v1.data_types import JsonObject + from sqlspec.adapters.spanner.type_converter import coerce_params_for_spanner, spanner_json, spanner_to_uuid from sqlspec.core import TypedParameter @@ -24,3 +28,77 @@ def test_coerce_params_unwraps_typed_datetime_parameter() -> None: coerced = coerce_params_for_spanner(params) assert coerced == {"available_at": timestamp} + + +def test_coerce_params_preserves_driver_ready_parameters() -> None: + timestamp = datetime(2026, 7, 4, 22, 9, 0, tzinfo=timezone.utc) + array = ["alpha", "beta"] + payload = JsonObject({"key": "value"}) + params = { + "id": 1, + "name": "alpha", + "enabled": True, + "score": 2.5, + "missing": None, + "day": date(2026, 7, 4), + "available_at": timestamp, + "tags": array, + "payload": payload, + } + + coerced = coerce_params_for_spanner(params) + + assert coerced is params + assert coerced["tags"] is array + assert coerced["payload"] is payload + + +def test_coerce_params_preserves_empty_parameter_mapping() -> None: + params: dict[str, object] = {} + + assert coerce_params_for_spanner(params) is params + + +def test_coerce_params_copies_only_when_values_require_conversion() -> None: + stdlib_uuid = UUID("550e8400-e29b-41d4-a716-446655440000") + utils_uuid = uuid_utils.UUID(str(stdlib_uuid)) + binary = b"binary" + naive_timestamp = datetime(2026, 7, 4, 22, 9, 0) + typed_timestamp = datetime(2026, 7, 5, 22, 9, 0, tzinfo=timezone.utc) + plain_array = ["alpha", "beta"] + params = { + "stdlib_uuid": stdlib_uuid, + "utils_uuid": utils_uuid, + "binary": binary, + "naive_timestamp": naive_timestamp, + "typed_timestamp": TypedParameter(typed_timestamp, datetime), + "payload": {"key": "value"}, + "tuple_array": ("alpha", "beta"), + "json_array": [{"key": "value"}], + "plain_array": plain_array, + } + + coerced = coerce_params_for_spanner(params) + + assert coerced is not params + assert coerced is not None + assert coerced["stdlib_uuid"] == base64.b64encode(stdlib_uuid.bytes) + assert coerced["utils_uuid"] == base64.b64encode(stdlib_uuid.bytes) + assert coerced["binary"] == base64.b64encode(binary) + assert coerced["naive_timestamp"] == naive_timestamp.replace(tzinfo=timezone.utc) + assert coerced["typed_timestamp"] is typed_timestamp + assert isinstance(coerced["payload"], JsonObject) + assert coerced["payload"] == {"key": "value"} + assert coerced["tuple_array"] == ["alpha", "beta"] + assert isinstance(coerced["json_array"], JsonObject) + assert coerced["json_array"].serialize() == '[{"key":"value"}]' + assert coerced["plain_array"] is plain_array + assert params["stdlib_uuid"] is stdlib_uuid + assert params["utils_uuid"] is utils_uuid + assert params["binary"] is binary + assert params["naive_timestamp"] is naive_timestamp + assert isinstance(params["typed_timestamp"], TypedParameter) + assert params["payload"] == {"key": "value"} + assert params["tuple_array"] == ("alpha", "beta") + assert params["json_array"] == [{"key": "value"}] + assert params["plain_array"] is plain_array From 3293c03afeb432937e3fab0e5df15407d8b73e2c Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 20:49:48 +0000 Subject: [PATCH 12/15] fix(bigquery): narrow validated record batches --- sqlspec/adapters/bigquery/driver.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sqlspec/adapters/bigquery/driver.py b/sqlspec/adapters/bigquery/driver.py index fd9edb761..f108028d2 100644 --- a/sqlspec/adapters/bigquery/driver.py +++ b/sqlspec/adapters/bigquery/driver.py @@ -793,9 +793,10 @@ def _records_to_json_rows( raise ImproperConfigurationError(msg) if type(record) is not dict or record_keys != resolved: can_reuse = False + validated_records = cast("list[Mapping[str, Any]]", materialized) if can_reuse: - return cast("list[dict[str, Any]]", materialized) - return [{column: record[column] for column in resolved} for record in materialized] + return cast("list[dict[str, Any]]", validated_records) + return [{column: record[column] for column in resolved} for record in validated_records] resolved = columns expected = set(resolved) From 04cd9011c59364ed296c9545b239299461e4470b Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 20:51:28 +0000 Subject: [PATCH 13/15] test(spanner): narrow optional coercion result --- tests/unit/adapters/test_spanner/test_type_converter.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/unit/adapters/test_spanner/test_type_converter.py b/tests/unit/adapters/test_spanner/test_type_converter.py index 662b4e6aa..efc9aeed5 100644 --- a/tests/unit/adapters/test_spanner/test_type_converter.py +++ b/tests/unit/adapters/test_spanner/test_type_converter.py @@ -48,6 +48,7 @@ def test_coerce_params_preserves_driver_ready_parameters() -> None: coerced = coerce_params_for_spanner(params) + assert coerced is not None assert coerced is params assert coerced["tags"] is array assert coerced["payload"] is payload From 27c98000838f5aa9e3b99232192721c116819c71 Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 21:15:59 +0000 Subject: [PATCH 14/15] test(spanner): type third-party JSON calls --- tests/unit/adapters/test_spanner/test_type_converter.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/unit/adapters/test_spanner/test_type_converter.py b/tests/unit/adapters/test_spanner/test_type_converter.py index efc9aeed5..1fc62d289 100644 --- a/tests/unit/adapters/test_spanner/test_type_converter.py +++ b/tests/unit/adapters/test_spanner/test_type_converter.py @@ -1,5 +1,6 @@ import base64 from datetime import date, datetime, timezone +from typing import Any, cast from uuid import UUID import uuid_utils @@ -33,7 +34,7 @@ def test_coerce_params_unwraps_typed_datetime_parameter() -> None: def test_coerce_params_preserves_driver_ready_parameters() -> None: timestamp = datetime(2026, 7, 4, 22, 9, 0, tzinfo=timezone.utc) array = ["alpha", "beta"] - payload = JsonObject({"key": "value"}) + payload = cast("Any", JsonObject)({"key": "value"}) params = { "id": 1, "name": "alpha", @@ -92,7 +93,7 @@ def test_coerce_params_copies_only_when_values_require_conversion() -> None: assert coerced["payload"] == {"key": "value"} assert coerced["tuple_array"] == ["alpha", "beta"] assert isinstance(coerced["json_array"], JsonObject) - assert coerced["json_array"].serialize() == '[{"key":"value"}]' + assert cast("Any", coerced["json_array"]).serialize() == '[{"key":"value"}]' assert coerced["plain_array"] is plain_array assert params["stdlib_uuid"] is stdlib_uuid assert params["utils_uuid"] is utils_uuid From e28a1678bbe60036f263d346a4a15f31ffcdb5ba Mon Sep 17 00:00:00 2001 From: Cody Fincher Date: Sat, 25 Jul 2026 23:58:12 +0000 Subject: [PATCH 15/15] chore(release): prepare v0.56.2 --- docs/changelog.rst | 44 +++++++++++++++++++++++++++++++++++++++----- pyproject.toml | 4 ++-- uv.lock | 14 +++++++------- 3 files changed, 48 insertions(+), 14 deletions(-) diff --git a/docs/changelog.rst b/docs/changelog.rst index d4b10baed..e7f6379fa 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -9,6 +9,45 @@ important operational fixes. Recent Updates ============== +v0.56.2 +------------------------------------------------------------------------------ + +**Added:** + +* Added ``uuid_from_string()``, ``uuid_from_bytes()``, and ``uuid_from_int()`` + in :mod:`sqlspec.utils.uuids`. They always return :class:`uuid.UUID`. Text + parsing uses Rust when ``uuid-utils`` is installed. + +**Fixed:** + +* PostgreSQL-family ADBC row APIs now decode scalar ``UUID`` and ``UUID[]`` data + to Python UUIDs. This works for buffered and streamed rows. Native Arrow + results keep the extension schema, and + ``enable_arrow_extension_types=False`` restores raw storage bytes on row + APIs. +* Oracle 12c through 20c can bind direct Python JSON values to ``BLOB IS JSON`` + columns. SQLSpec writes UTF-8 JSON to a BLOB locator for sync, async, batch, + and streaming calls. Oracle 21c and newer still use native ``JSON`` binding. + Explicit ``OracleClob`` values remain CLOBs. + +**Changed:** + +* BigQuery ``load_from_records()`` now reuses fully checked lists of plain + dictionaries when no fields must move. It still copies rows for explicit + columns, mapping subclasses, and different key orders. +* Spanner and Oracle now reuse bind data when no value needs a conversion. They + copy only after the first changed value. Bound values and checks are + unchanged. +* UUID parsing now uses :mod:`sqlspec.utils.uuids` across adapters and type + converters. + +**Docs:** + +* The Oracle guide now covers JSON storage, LOB, UUID, and VECTOR behavior. It + also covers driver options and an Oracle ``MERGE`` upsert recipe. +* The ADBC guide now explains UUID row and Arrow results. It also documents the + ``enable_arrow_extension_types`` switch. + v0.56.1 ------------------------------------------------------------------------------ @@ -21,11 +60,6 @@ v0.56.1 **Fixed:** -* PostgreSQL-family ADBC row APIs now decode scalar ``UUID`` and ``UUID[]`` - opaque storage values to Python UUIDs across buffered and streamed results. - Native Arrow results preserve the extension schema, and - ``enable_arrow_extension_types=False`` restores raw storage bytes on row - APIs. * PostgreSQL-family ADBC drivers no longer fail when a UUID is a statement's only parameter. Repeated executions of such a statement previously reused a cached plan that skipped UUID binding and reached PostgreSQL as ``bytea``. diff --git a/pyproject.toml b/pyproject.toml index d1bc331b4..5e276ee06 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ maintainers = [{ name = "Litestar Developers", email = "hello@litestar.dev" }] name = "sqlspec" readme = "README.md" requires-python = ">=3.10, <4.0" -version = "0.56.1" +version = "0.56.2" [project.urls] Discord = "https://discord.gg/litestar" @@ -308,7 +308,7 @@ opt_level = "3" # Maximum optimization (0-3) allow_dirty = true commit = false commit_args = "--no-verify" -current_version = "0.56.1" +current_version = "0.56.2" ignore_missing_files = false ignore_missing_version = false message = "chore(release): bump to v{new_version}" diff --git a/uv.lock b/uv.lock index 0e2a42eff..43e71ea83 100644 --- a/uv.lock +++ b/uv.lock @@ -1529,11 +1529,11 @@ wheels = [ [[package]] name = "fastjsonschema" -version = "2.21.2" +version = "2.22.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/20/b5/23b216d9d985a956623b6bd12d4086b60f0059b27799f23016af04a74ea1/fastjsonschema-2.21.2.tar.gz", hash = "sha256:b1eb43748041c880796cd077f1a07c3d94e93ae84bba5ed36800a33554ae05de", size = 374130, upload-time = "2025-08-14T18:49:36.666Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f1/11/c802f752919c1fd83d44b3b955c6e7120c79f355d4a448c358198a11178c/fastjsonschema-2.22.0.tar.gz", hash = "sha256:6eb12e8f9900db6166c3d396d178ebdf6a4215fe22a06e19792edd612a20035a", size = 382291, upload-time = "2026-07-25T20:32:35.561Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/cb/a8/20d0723294217e47de6d9e2e40fd4a9d2f7c4b6ef974babd482a59743694/fastjsonschema-2.21.2-py3-none-any.whl", hash = "sha256:1c797122d0a86c5cace2e54bf4e819c36223b552017172f32c5c024a6b77e463", size = 24024, upload-time = "2025-08-14T18:49:34.776Z" }, + { url = "https://files.pythonhosted.org/packages/a4/9d/4a0f9355ca3e540b2d22d5f269212e2f227b8f277835b02d8908355245d1/fastjsonschema-2.22.0-py3-none-any.whl", hash = "sha256:60f4c92fda6f93efe3b3261638836478e1e11abc01c647e36e478199f7a86a37", size = 26248, upload-time = "2026-07-25T20:32:33.616Z" }, ] [[package]] @@ -5395,11 +5395,11 @@ wheels = [ [[package]] name = "pytz" -version = "2026.2" +version = "2026.3.post1" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/ff/46/dd499ec9038423421951e4fad73051febaa13d2df82b4064f87af8b8c0c3/pytz-2026.2.tar.gz", hash = "sha256:0e60b47b29f21574376f218fe21abc009894a2321ea16c6754f3cad6eb7cdd6a", size = 320861, upload-time = "2026-05-04T01:35:29.667Z" } +sdist = { url = "https://files.pythonhosted.org/packages/fb/48/fb042503b6ca6cd271261dc559fd6432f7d8c713153e9ec5c591af4dfc1c/pytz-2026.3.post1.tar.gz", hash = "sha256:2211d3fcf9a797d3405cac96ac7f61d80e6a644f72a3309607282fe8a2010c5d", size = 319745, upload-time = "2026-07-25T15:12:07.385Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ec/dd/96da98f892250475bdf2328112d7468abdd4acc7b902b6af23f4ed958ea0/pytz-2026.2-py2.py3-none-any.whl", hash = "sha256:04156e608bee23d3792fd45c94ae47fae1036688e75032eea2e3bf0323d1f126", size = 510141, upload-time = "2026-05-04T01:35:27.408Z" }, + { url = "https://files.pythonhosted.org/packages/0f/7b/39c34ca613b0b198cb866466651b26b045e2009864c5183c979a3b83f383/pytz-2026.3.post1-py2.py3-none-any.whl", hash = "sha256:dd95840dd199baea12d9cc096a1d452caa6596a1c1e4b5f3dbd1541855d5e815", size = 508283, upload-time = "2026-07-25T15:12:05.782Z" }, ] [[package]] @@ -6636,7 +6636,7 @@ wheels = [ [[package]] name = "sqlspec" -version = "0.56.1" +version = "0.56.2" source = { editable = "." } dependencies = [ { name = "mypy-extensions" },