From 681f3de038547e8706b311102276dae1d425779b Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Mon, 27 Jul 2026 17:48:16 +0000 Subject: [PATCH 1/3] feat(sqlalchemy-spanner): native UUID data type support in dialect --- .../sqlalchemy_spanner/sqlalchemy_spanner.py | 16 ++++ .../tests/mockserver_tests/test_uuid.py | 88 +++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py diff --git a/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py b/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py index 1a303e630d41..5807202ff32c 100644 --- a/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py +++ b/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py @@ -123,6 +123,9 @@ def process(value): "TOKENLIST": types.String, } +if hasattr(types, "UUID"): + _type_map["UUID"] = types.UUID + _type_map_inv = { types.Boolean: "BOOL", @@ -141,6 +144,12 @@ def process(value): types.NullType: "INT64", } +if hasattr(types, "UUID"): + _type_map_inv[types.UUID] = "UUID" + +if hasattr(types, "Uuid"): + _type_map_inv[types.Uuid] = "UUID" + _compound_keywords = { selectable.CompoundSelect.UNION: "UNION DISTINCT", selectable.CompoundSelect.UNION_ALL: "UNION ALL", @@ -765,6 +774,12 @@ class SpannerTypeCompiler(GenericTypeCompiler): Maps SQLAlchemy types to Spanner data types. """ + def visit_uuid(self, type_, **kw): + if not type_.native_uuid or not self.dialect.supports_native_uuid: + return self.visit_CHAR(types.CHAR(36), **kw) + else: + return "UUID" + def visit_INTEGER(self, type_, **kw): return "INT64" @@ -847,6 +862,7 @@ class SpannerDialect(DefaultDialect): supports_identity_columns = True supports_native_boolean = True supports_native_decimal = True + supports_native_uuid = True supports_statement_cache = True # Spanner uses protos for enums. Creating a column like # Column("an_enum", Enum("A", "B", "C")) will result in a String diff --git a/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py new file mode 100644 index 000000000000..d97c4bb3e50f --- /dev/null +++ b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py @@ -0,0 +1,88 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from sqlalchemy import Column, MetaData, Table, types +from sqlalchemy.schema import CreateTable +from sqlalchemy.testing import eq_, fixtures +from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import ( + SpannerDialect, + _type_map, + _type_map_inv, +) + + +class UuidTest(fixtures.TestBase): + def test_uuid_type_mapping(self): + """Test UUID is registered in _type_map and _type_map_inv.""" + assert "UUID" in _type_map + eq_(_type_map["UUID"], types.UUID) + assert types.UUID in _type_map_inv + eq_(_type_map_inv[types.UUID], "UUID") + assert types.Uuid in _type_map_inv + eq_(_type_map_inv[types.Uuid], "UUID") + + def test_uuid_designate_type(self): + """Test reflecting UUID type string returns types.UUID.""" + dialect = SpannerDialect() + assert dialect.supports_native_uuid is True + col_type = dialect._designate_type("UUID") + eq_(col_type, types.UUID) + + def test_uuid_ddl_compilation_default(self): + """Test DDL compilation emits UUID type by default when supports_native_uuid + is True. + """ + dialect = SpannerDialect() + metadata = MetaData() + table = Table( + "test_uuid_table", + metadata, + Column("user_id", types.Uuid, primary_key=True), + ) + statement = str(CreateTable(table).compile(dialect=dialect)).strip() + assert "user_id UUID NOT NULL" in statement + + def test_uuid_ddl_compilation_native_disabled(self): + """Test DDL compilation falls back to STRING(36) when supports_native_uuid + is False. + """ + dialect = SpannerDialect() + dialect.supports_native_uuid = False + metadata = MetaData() + table = Table( + "test_uuid_table", + metadata, + Column("user_id", types.Uuid, primary_key=True), + ) + statement = str(CreateTable(table).compile(dialect=dialect)).strip() + assert "user_id STRING(36) NOT NULL" in statement + + def test_uuid_python_conversion_legacy(self): + """Test that types.Uuid automatically converts uuid.UUID to/from str + when native_uuid is False. + """ + import uuid + + dialect = SpannerDialect() + dialect.supports_native_uuid = False + uuid_type = types.Uuid() + test_uuid = uuid.UUID("123e4567-e89b-12d3-a456-426614174000") + + # Test bind processor converts uuid.UUID -> str + bind_proc = uuid_type.bind_processor(dialect) + eq_(bind_proc(test_uuid), "123e4567e89b12d3a456426614174000") + + # Test result processor converts str -> uuid.UUID + res_proc = uuid_type.result_processor(dialect, "STRING") + eq_(res_proc("123e4567-e89b-12d3-a456-426614174000"), test_uuid) From b60699e0e6603fd4bc8e3529ede9452bcac69c5e Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 28 Jul 2026 06:51:24 +0000 Subject: [PATCH 2/3] fix(sqlalchemy-spanner): default supports_native_uuid to False for legacy Uuid compatibility --- .../sqlalchemy_spanner/sqlalchemy_spanner.py | 4 ++-- .../tests/mockserver_tests/test_uuid.py | 22 ++++++++++--------- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py b/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py index 5807202ff32c..a7bf749eff2d 100644 --- a/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py +++ b/packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py @@ -778,7 +778,7 @@ def visit_uuid(self, type_, **kw): if not type_.native_uuid or not self.dialect.supports_native_uuid: return self.visit_CHAR(types.CHAR(36), **kw) else: - return "UUID" + return self.visit_UUID(type_, **kw) def visit_INTEGER(self, type_, **kw): return "INT64" @@ -862,7 +862,7 @@ class SpannerDialect(DefaultDialect): supports_identity_columns = True supports_native_boolean = True supports_native_decimal = True - supports_native_uuid = True + supports_native_uuid = False supports_statement_cache = True # Spanner uses protos for enums. Creating a column like # Column("an_enum", Enum("A", "B", "C")) will result in a String diff --git a/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py index d97c4bb3e50f..9cb1ba07cabe 100644 --- a/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py +++ b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py @@ -35,30 +35,32 @@ def test_uuid_type_mapping(self): def test_uuid_designate_type(self): """Test reflecting UUID type string returns types.UUID.""" dialect = SpannerDialect() - assert dialect.supports_native_uuid is True + assert dialect.supports_native_uuid is False col_type = dialect._designate_type("UUID") eq_(col_type, types.UUID) def test_uuid_ddl_compilation_default(self): - """Test DDL compilation emits UUID type by default when supports_native_uuid - is True. + """Test DDL compilation: types.Uuid emits STRING(36) by default and + types.UUID emits UUID by default. """ dialect = SpannerDialect() metadata = MetaData() table = Table( "test_uuid_table", metadata, - Column("user_id", types.Uuid, primary_key=True), + Column("legacy_id", types.Uuid, primary_key=True), + Column("native_id", types.UUID), ) statement = str(CreateTable(table).compile(dialect=dialect)).strip() - assert "user_id UUID NOT NULL" in statement + assert "legacy_id STRING(36) NOT NULL" in statement + assert "native_id UUID" in statement - def test_uuid_ddl_compilation_native_disabled(self): - """Test DDL compilation falls back to STRING(36) when supports_native_uuid - is False. + def test_uuid_ddl_compilation_native_enabled(self): + """Test DDL compilation emits UUID type for types.Uuid when supports_native_uuid + is set to True on dialect. """ dialect = SpannerDialect() - dialect.supports_native_uuid = False + dialect.supports_native_uuid = True metadata = MetaData() table = Table( "test_uuid_table", @@ -66,7 +68,7 @@ def test_uuid_ddl_compilation_native_disabled(self): Column("user_id", types.Uuid, primary_key=True), ) statement = str(CreateTable(table).compile(dialect=dialect)).strip() - assert "user_id STRING(36) NOT NULL" in statement + assert "user_id UUID NOT NULL" in statement def test_uuid_python_conversion_legacy(self): """Test that types.Uuid automatically converts uuid.UUID to/from str From b17241d17073d55696fe6a426f27cce4a7037afb Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 28 Jul 2026 10:01:46 +0000 Subject: [PATCH 3/3] test(sqlalchemy-spanner): add mockserver 15-matrix tests and sample for types.Uuid vs types.UUID --- .../sqlalchemy-spanner/samples/uuid_sample.py | 187 +++++++++++++++ .../tests/mockserver_tests/test_uuid.py | 224 ++++++++++++++++-- 2 files changed, 392 insertions(+), 19 deletions(-) create mode 100644 packages/sqlalchemy-spanner/samples/uuid_sample.py diff --git a/packages/sqlalchemy-spanner/samples/uuid_sample.py b/packages/sqlalchemy-spanner/samples/uuid_sample.py new file mode 100644 index 000000000000..27d725b56cb8 --- /dev/null +++ b/packages/sqlalchemy-spanner/samples/uuid_sample.py @@ -0,0 +1,187 @@ +# Copyright 2026 Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +from uuid import UUID, uuid4 + +from sample_helper import run_sample +from sqlalchemy import create_engine, select, types +from sqlalchemy.dialects import registry +from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column + +registry.register( + "spanner", + "google.cloud.sqlalchemy_spanner.sqlalchemy_spanner", + "SpannerDialect", +) + + +class Base(DeclarativeBase): + pass + + +# ----------------------------------------------------------------------------- +# GUIDELINES FOR USERS / CUSTOMERS: +# +# 1. types.Uuid() (CamelCase - Recommended): +# - supports_native_uuid = False (default): Compiles to `STRING(36)`. +# - supports_native_uuid = True: Compiles to `UUID`. +# - Works seamlessly for inserts and queries across both configurations. +# +# 2. types.UUID() (All-Caps - Native Spanner Type): +# - IMPORTANT: Use `types.UUID()` ONLY when `supports_native_uuid = True` +# is enabled on your dialect/engine for native Cloud Spanner UUID columns. +# +# 3. types.Uuid(native_uuid=False) (Explicit Override): +# - Forces `STRING(36)` storage even when `supports_native_uuid = True`. +# ----------------------------------------------------------------------------- + + +class CustomerLegacy(Base): + """Uses types.Uuid() -> Compiles to STRING(36) by default.""" + + __tablename__ = "customers_legacy" + id: Mapped[UUID] = mapped_column(types.Uuid(), primary_key=True, default=uuid4) + name: Mapped[str] = mapped_column() + + +class CustomerNative(Base): + """Uses types.UUID() -> ALWAYS compiles to native UUID type in Spanner DDL, + + even when supports_native_uuid=False. + """ + + __tablename__ = "customers_native" + id: Mapped[UUID] = mapped_column(types.UUID(), primary_key=True, default=uuid4) + name: Mapped[str] = mapped_column() + + +class CustomerExplicitString(Base): + """Uses types.Uuid(native_uuid=False) -> ALWAYS compiles to STRING(36).""" + + __tablename__ = "customers_explicit_string" + id: Mapped[UUID] = mapped_column( + types.Uuid(native_uuid=False), + primary_key=True, + default=uuid4, + ) + name: Mapped[str] = mapped_column() + + +def run_with_flag_disabled(): + """Scenario 1: supports_native_uuid = False (Default). + + - types.Uuid() emits STRING(36) in DDL. + - types.Uuid(native_uuid=False) emits STRING(36) in DDL. + """ + sep = "=" * 73 + print(f"\n{sep}") + print("SCENARIO 1: Flag Disabled (supports_native_uuid=False)") + print(sep) + engine = create_engine( + "spanner:///projects/sample-project/" + "instances/sample-instance/" + "databases/sample-database", + echo=True, + ) + engine.dialect.supports_native_uuid = False + Base.metadata.create_all(engine) + + with Session(engine) as session: + cust_legacy = CustomerLegacy(name="Alice_Legacy_FlagDisabled") + cust_override = CustomerExplicitString(name="Diana_String_FlagDisabled") + session.add_all([cust_legacy, cust_override]) + session.commit() + + fetched_legacy = session.scalars( + select(CustomerLegacy).filter_by(id=cust_legacy.id) + ).one() + print( + "--> [Flag Disabled] Legacy Customer ID " + f"({type(fetched_legacy.id).__name__}): {fetched_legacy.id}" + ) + + fetched_override = session.scalars( + select(CustomerExplicitString).filter_by(id=cust_override.id) + ).one() + print( + "--> [Flag Disabled] Explicit STRING Customer ID " + f"({type(fetched_override.id).__name__}): {fetched_override.id}" + ) + + +def run_with_flag_enabled(): + """Scenario 2: supports_native_uuid = True. + + - types.Uuid() emits UUID in DDL when flag enabled. + - types.UUID() ALWAYS emits UUID in DDL. + - types.Uuid(native_uuid=False) explicitly emits STRING(36). + """ + sep = "=" * 73 + print(f"\n{sep}") + print("SCENARIO 2: Flag Enabled (supports_native_uuid=True)") + print(sep) + engine = create_engine( + "spanner:///projects/sample-project/" + "instances/sample-instance/" + "databases/sample-database", + echo=True, + ) + engine.dialect.supports_native_uuid = True + Base.metadata.create_all(engine) + + with Session(engine) as session: + cust_legacy = CustomerLegacy(name="Charlie_Legacy_FlagEnabled") + cust_native = CustomerNative(name="Bob_Native_FlagEnabled") + cust_override = CustomerExplicitString(name="Diana_String_FlagEnabled") + session.add_all([cust_legacy, cust_native, cust_override]) + session.commit() + + fetched_legacy = session.scalars( + select(CustomerLegacy).filter_by(id=cust_legacy.id) + ).one() + print( + "--> [Flag Enabled] Legacy Customer ID " + f"({type(fetched_legacy.id).__name__}): {fetched_legacy.id}" + ) + + fetched_native = session.scalars( + select(CustomerNative).filter_by(id=cust_native.id) + ).one() + print( + "--> [Flag Enabled] Native Customer ID " + f"({type(fetched_native.id).__name__}): {fetched_native.id}" + ) + + fetched_override = session.scalars( + select(CustomerExplicitString).filter_by(id=cust_override.id) + ).one() + print( + "--> [Flag Enabled] Explicit STRING Customer ID " + f"({type(fetched_override.id).__name__}): {fetched_override.id}" + ) + + +def uuid_sample(): + """Demonstrates how to use types.Uuid vs types.UUID with Spanner SQLAlchemy + + under both supports_native_uuid=False and supports_native_uuid=True. + """ + run_with_flag_disabled() + run_with_flag_enabled() + + +if __name__ == "__main__": + run_sample(uuid_sample) diff --git a/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py index 9cb1ba07cabe..d330f6e48ad5 100644 --- a/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py +++ b/packages/sqlalchemy-spanner/tests/mockserver_tests/test_uuid.py @@ -12,17 +12,48 @@ # See the License for the specific language governing permissions and # limitations under the License. -from sqlalchemy import Column, MetaData, Table, types +from uuid import UUID + + +from google.cloud.spanner_v1 import TypeCode +from sqlalchemy import Column, MetaData, Table, select, types +from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column from sqlalchemy.schema import CreateTable -from sqlalchemy.testing import eq_, fixtures +from sqlalchemy.testing import eq_ + from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import ( SpannerDialect, _type_map, _type_map_inv, ) +from tests.mockserver_tests.mock_server_test_base import ( + MockServerTestBase, + add_single_result, +) + +class TestUuidMockServer(MockServerTestBase): + def _get_models(self): + class Base(DeclarativeBase): + pass + + class UserUuid(Base): + __tablename__ = "users_uuid" + id: Mapped[UUID] = mapped_column(types.Uuid(), primary_key=True) + + class UserUUID(Base): + __tablename__ = "users_UUID" + id: Mapped[UUID] = mapped_column(types.UUID(), primary_key=True) + + class UserUuidNativeFalse(Base): + __tablename__ = "users_uuid_native_false" + id: Mapped[UUID] = mapped_column( + types.Uuid(native_uuid=False), + primary_key=True, + ) + + return UserUuid, UserUUID, UserUuidNativeFalse -class UuidTest(fixtures.TestBase): def test_uuid_type_mapping(self): """Test UUID is registered in _type_map and _type_map_inv.""" assert "UUID" in _type_map @@ -56,8 +87,8 @@ def test_uuid_ddl_compilation_default(self): assert "native_id UUID" in statement def test_uuid_ddl_compilation_native_enabled(self): - """Test DDL compilation emits UUID type for types.Uuid when supports_native_uuid - is set to True on dialect. + """Test DDL compilation emits UUID type for types.Uuid when + supports_native_uuid is set to True on dialect. """ dialect = SpannerDialect() dialect.supports_native_uuid = True @@ -70,21 +101,176 @@ def test_uuid_ddl_compilation_native_enabled(self): statement = str(CreateTable(table).compile(dialect=dialect)).strip() assert "user_id UUID NOT NULL" in statement - def test_uuid_python_conversion_legacy(self): - """Test that types.Uuid automatically converts uuid.UUID to/from str - when native_uuid is False. + def test_uuid_ddl_compilation_native_false_override(self): + """Test DDL compilation emits STRING(36) for native_uuid=False + even when supports_native_uuid is set to True on dialect. """ - import uuid - dialect = SpannerDialect() - dialect.supports_native_uuid = False - uuid_type = types.Uuid() - test_uuid = uuid.UUID("123e4567-e89b-12d3-a456-426614174000") + dialect.supports_native_uuid = True + metadata = MetaData() + table = Table( + "test_uuid_table", + metadata, + Column("user_id", types.Uuid(native_uuid=False), primary_key=True), + ) + statement = str(CreateTable(table).compile(dialect=dialect)).strip() + assert "user_id STRING(36) NOT NULL" in statement + + # ----------------------------------------------------------------- + # 1. Flag is disabled (supports_native_uuid = False) + # ----------------------------------------------------------------- + def test_1_1_flag_disabled_uuid_db_string(self): + """1.1 Flag disabled + mapped Uuid + DB STRING(36) -> uuid.UUID""" + UserUuid, _, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = False + sql = "SELECT users_uuid.id\nFROM users_uuid" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUuid.id)).first() + eq_(type(user_id), UUID) + eq_(user_id, UUID(raw_uuid_str)) - # Test bind processor converts uuid.UUID -> str - bind_proc = uuid_type.bind_processor(dialect) - eq_(bind_proc(test_uuid), "123e4567e89b12d3a456426614174000") + def test_1_2_flag_disabled_uuid_db_uuid(self): + """1.2 Flag disabled + mapped Uuid + DB UUID -> AttributeError""" + UserUuid, _, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = False + sql = "SELECT users_uuid.id\nFROM users_uuid" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)]) + + with Session(engine) as session: + try: + _ = session.scalars(select(UserUuid.id)).first() + assert False, "Expected AttributeError" + except AttributeError: + pass + + def test_1_3_flag_disabled_UUID_db_string(self): + """1.3 Flag disabled + mapped UUID + DB STRING(36) -> uuid.UUID""" + _, UserUUID, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = False + sql = "SELECT `users_UUID`.id\nFROM `users_UUID`" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUUID.id)).first() + eq_(type(user_id), UUID) + eq_(user_id, UUID(raw_uuid_str)) + + def test_1_4_flag_disabled_UUID_db_uuid(self): + """1.4 Flag disabled + mapped UUID + DB UUID -> AttributeError""" + _, UserUUID, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = False + sql = "SELECT `users_UUID`.id\nFROM `users_UUID`" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)]) + + with Session(engine) as session: + try: + _ = session.scalars(select(UserUUID.id)).first() + assert False, "Expected AttributeError" + except AttributeError: + pass + + # ----------------------------------------------------------------- + # 2. Flag is enabled (supports_native_uuid = True) + # ----------------------------------------------------------------- + def test_2_1_flag_enabled_uuid_db_string(self): + """2.1 Flag enabled + mapped Uuid + DB STRING(36) -> user.id is str""" + UserUuid, _, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT users_uuid.id\nFROM users_uuid" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUuid.id)).first() + eq_(type(user_id), str) + eq_(user_id, raw_uuid_str) + + def test_2_2_flag_enabled_uuid_db_uuid(self): + """2.2 Flag enabled + mapped Uuid + DB UUID -> user.id is uuid.UUID""" + UserUuid, _, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT users_uuid.id\nFROM users_uuid" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUuid.id)).first() + eq_(type(user_id), UUID) + eq_(user_id, UUID(raw_uuid_str)) + + def test_2_3_flag_enabled_UUID_db_string(self): + """2.3 Flag enabled + mapped UUID (all-caps) + DB STRING(36) -> str""" + _, UserUUID, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT `users_UUID`.id\nFROM `users_UUID`" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUUID.id)).first() + eq_(type(user_id), str) + eq_(user_id, raw_uuid_str) + + def test_2_4_flag_enabled_UUID_db_uuid(self): + """2.4 Flag enabled + mapped UUID (all-caps) + DB UUID -> uuid.UUID""" + _, UserUUID, _ = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT `users_UUID`.id\nFROM `users_UUID`" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUUID.id)).first() + eq_(type(user_id), UUID) + eq_(user_id, UUID(raw_uuid_str)) + + # ----------------------------------------------------------------- + # 3. Explicit native_uuid = False override + # ----------------------------------------------------------------- + def test_3_1_flag_enabled_native_false_override_db_string(self): + """3.1 Flag enabled + types.Uuid(native_uuid=False) + DB STRING(36) + -> user.id is uuid.UUID (override forces conversion) + """ + _, _, UserUuidNativeFalse = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT users_uuid_native_false.id\nFROM users_uuid_native_false" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.STRING, [(raw_uuid_str,)]) + + with Session(engine) as session: + user_id = session.scalars(select(UserUuidNativeFalse.id)).first() + eq_(type(user_id), UUID) + eq_(user_id, UUID(raw_uuid_str)) + + def test_3_2_flag_enabled_native_false_override_db_uuid(self): + """3.2 Flag enabled + types.Uuid(native_uuid=False) + DB UUID + -> AttributeError (override forces conversion, failing on native UUID) + """ + _, _, UserUuidNativeFalse = self._get_models() + engine = self.create_engine() + engine.dialect.supports_native_uuid = True + sql = "SELECT users_uuid_native_false.id\nFROM users_uuid_native_false" + raw_uuid_str = "550e8400-e29b-41d4-a716-446655440000" + add_single_result(sql, "id", TypeCode.UUID, [(raw_uuid_str,)]) - # Test result processor converts str -> uuid.UUID - res_proc = uuid_type.result_processor(dialect, "STRING") - eq_(res_proc("123e4567-e89b-12d3-a456-426614174000"), test_uuid) + with Session(engine) as session: + try: + _ = session.scalars(select(UserUuidNativeFalse.id)).first() + assert False, "Expected AttributeError" + except AttributeError: + pass