fix(spanner, sqlalchemy-spanner): fix reflection crashes, add native UUID support, and improve JsonObject

- Exclude SEARCH indexes and guard None column_sorting in SpannerDialect.get_multi_indexes to prevent reflection AttributeError crashes.
- Register TOKENLIST in _type_map to enable table reflection for TOKENLIST columns without KeyError.
- Add native UUID support in SpannerDialect (_type_map, _type_map_inv, SpannerDDLCompiler.visit_UUID, SpannerTypeCompiler.visit_UUID/visit_uuid) while preserving STRING(36) backward compatibility.
- Fix spanner_storing column resolution in SpannerDDLCompiler.visit_create_index for unbound columns in Alembic batch mode.
- Add to_python() method and public properties (is_null, is_array, is_scalar) to JsonObject in google-cloud-spanner.
- Add unit tests in test_dialect.py and mockserver integration tests in test_dialect_integration.py.

Author: sakthivelmanii

packages/google-cloud-spanner/google/cloud/spanner_v1/data_types.py

@@ -57,6 +57,37 @@ def __init__(self, *args, **kwargs):
         if not self._is_null:
             super(JsonObject, self).__init__(*args, **kwargs)
 
+    @property
+    def is_null(self):
+        """Return True if JsonObject represents JSON null."""
+        return self._is_null
+
+    @property
+    def is_array(self):
+        """Return True if JsonObject represents a JSON array."""
+        return self._is_array
+
+    @property
+    def is_scalar(self):
+        """Return True if JsonObject represents a JSON scalar value."""
+        return self._is_scalar_value
+
+    def to_python(self):
+        """Return unwrapped native Python object representation (dict, list, scalar, or None)."""
+        if self._is_null:
+            return None
+        if self._is_array:
+            return [
+                item.to_python() if isinstance(item, JsonObject) else item
+                for item in self._array_value
+            ]
+        if self._is_scalar_value:
+            return self._simple_value
+        return {
+            k: (v.to_python() if isinstance(v, JsonObject) else v)
+            for k, v in self.items()
+        }
+
     def __repr__(self):
         if self._is_array:
             return str(self._array_value)

packages/google-cloud-spanner/tests/unit/test_datatypes.py

@@ -96,3 +96,26 @@ def test_w_JsonObject_of_list_of_simple_JsonData(self):
         expected = json.dumps(data, sort_keys=True, separators=(",", ":"))
         data_jsonobject = JsonObject(JsonObject(data))
         self.assertEqual(data_jsonobject.serialize(), expected)
+
+    def test_to_python_dict(self):
+        obj = JsonObject({"a": 1, "b": [2, 3]})
+        self.assertFalse(obj.is_null)
+        self.assertFalse(obj.is_array)
+        self.assertFalse(obj.is_scalar)
+        self.assertEqual(obj.to_python(), {"a": 1, "b": [2, 3]})
+
+    def test_to_python_array(self):
+        obj = JsonObject([{"a": 1}, 2, "str"])
+        self.assertFalse(obj.is_null)
+        self.assertTrue(obj.is_array)
+        self.assertFalse(obj.is_scalar)
+        self.assertEqual(obj.to_python(), [{"a": 1}, 2, "str"])
+
+    def test_to_python_scalar_and_null(self):
+        scalar_obj = JsonObject("hello")
+        self.assertTrue(scalar_obj.is_scalar)
+        self.assertEqual(scalar_obj.to_python(), "hello")
+
+        null_obj = JsonObject(None)
+        self.assertTrue(null_obj.is_null)
+        self.assertIsNone(null_obj.to_python())

packages/sqlalchemy-spanner/.coveragerc

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+#
+# 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
+#
+#     https://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.
+
+[run]
+branch = True
+source =
+    google/cloud/sqlalchemy_spanner
+
+[paths]
+source =
+    google/cloud/sqlalchemy_spanner
+    */site-packages/google/cloud/sqlalchemy_spanner
+
+[report]
+show_missing = True
+exclude_lines =
+    pragma: NO COVER
+    def __repr__
+    raise NotImplementedError
+omit =
+  tests/*
+  */tests/*

packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -120,6 +120,8 @@ def process(value):
     "TIMESTAMP": types.TIMESTAMP,
     "ARRAY": types.ARRAY,
     "JSON": types.JSON,
+    "TOKENLIST": types.String,
+    "UUID": types.UUID,
 }
 
 
@@ -136,6 +138,7 @@ def process(value):
     types.String: "STRING",
     types.TIME: "TIME",
     types.TIMESTAMP: "TIMESTAMP",
+    types.UUID: "UUID",
     types.Integer: "INT64",
     types.NullType: "INT64",
 }
@@ -426,8 +429,10 @@ def returning_clause(self, stmt, returning_cols, **kw):
             )
             for c in expression._select_iterables(
                 filter(
-                    lambda col: not col.dialect_options.get("spanner", {}).get(
-                        "exclude_from_returning", False
+                    lambda col: (
+                        not col.dialect_options.get("spanner", {}).get(
+                            "exclude_from_returning", False
+                        )
                     ),
                     returning_cols,
                 )
@@ -709,12 +714,16 @@ def visit_create_index(
             options = index.dialect_options["spanner"]
             if "storing" in options:
                 storing = options["storing"]
-                storing_columns = [
-                    index.table.c[col] if isinstance(col, str) else col
-                    for col in storing
-                ]
+                storing_names = []
+                for col in storing:
+                    if isinstance(col, str):
+                        storing_names.append(col)
+                    elif hasattr(col, "name"):
+                        storing_names.append(col.name)
+                    else:
+                        storing_names.append(str(col))
                 text += " STORING (%s)" % ", ".join(
-                    [self.preparer.quote(c.name) for c in storing_columns]
+                    [self.preparer.quote(name) for name in storing_names]
                 )
 
             interleave_in = options.get("interleave_in")
@@ -814,6 +823,12 @@ def visit_NUMERIC(self, type_, **kw):
     def visit_BIGINT(self, type_, **kw):
         return "INT64"
 
+    def visit_UUID(self, type_, **kw):
+        return "UUID"
+
+    def visit_uuid(self, type_, **kw):
+        return "UUID"
+
     def visit_JSON(self, type_, **kw):
         return "JSON"
 
@@ -1300,6 +1315,7 @@ def get_multi_indexes(
                 {table_type_query}
                 {schema_filter_query}
                 i.index_type != 'PRIMARY_KEY'
+                AND i.index_type != 'SEARCH'
                 AND i.spanner_is_managed = FALSE
             GROUP BY i.table_catalog, i.table_schema, i.table_name,
                      i.index_name, i.is_unique
@@ -1324,7 +1340,8 @@ def get_multi_indexes(
                     "column_names": row[3],
                     "unique": row[4],
                     "column_sorting": {
-                        col: order.lower() for col, order in zip(row[3], row[5])
+                        col: (order.lower() if order else None)
+                        for col, order in zip(row[3], row[5] or [])
                     },
                     "include_columns": include_columns if include_columns else [],
                     "dialect_options": dialect_options,

packages/sqlalchemy-spanner/noxfile.py

@@ -85,6 +85,7 @@ class = StreamHandler
 UNIT_TEST_STANDARD_DEPENDENCIES = [
     "mock",
     "pytest",
+    "pytest-cov",
 ]
 
 UNIT_TEST_EXTERNAL_DEPENDENCIES = [
@@ -275,6 +276,9 @@ def mockserver(session):
     session.run(
         "py.test",
         "--quiet",
+        "--cov=google.cloud.sqlalchemy_spanner",
+        "--cov-append",
+        "--cov-config=.coveragerc",
         os.path.join("tests", "mockserver_tests"),
         *session.posargs,
     )
@@ -368,7 +372,15 @@ def unit(session, test_type):
             *UNIT_TEST_DEPENDENCIES,
         )
         session.install(".")
-        session.run("py.test", "--quiet", os.path.join("tests/unit"), *session.posargs)
+        session.run(
+            "py.test",
+            "--quiet",
+            "--cov=google.cloud.sqlalchemy_spanner",
+            "--cov-append",
+            "--cov-config=.coveragerc",
+            os.path.join("tests/unit"),
+            *session.posargs,
+        )
         return
 
 

packages/sqlalchemy-spanner/tests/mockserver_tests/test_dialect_integration.py

@@ -0,0 +1,83 @@
+# 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 google.cloud.spanner_admin_database_v1 import UpdateDatabaseDdlRequest
+from google.cloud.spanner_v1 import ResultSet
+from sqlalchemy import Column, Index, MetaData, Table, Uuid, types
+from sqlalchemy.testing import eq_, is_instance_of
+
+from tests.mockserver_tests.mock_server_test_base import (
+    MockServerTestBase,
+    add_result,
+)
+
+
+class TestDialectIntegration(MockServerTestBase):
+    def test_create_table_with_native_uuid(self):
+        """Integration test verifying native UUID and TOKENLIST DDL generation."""
+        add_result(
+            """SELECT true
+FROM INFORMATION_SCHEMA.TABLES
+WHERE TABLE_SCHEMA="" AND TABLE_NAME="products"
+LIMIT 1
+""",
+            ResultSet(),
+        )
+        engine = self.create_engine()
+        metadata = MetaData()
+        Table(
+            "products",
+            metadata,
+            Column("product_id", Uuid, primary_key=True),
+            Column("token_data", types.String()),
+        )
+        metadata.create_all(engine)
+        requests = self.database_admin_service.requests
+        eq_(1, len(requests))
+        is_instance_of(requests[0], UpdateDatabaseDdlRequest)
+        statement = requests[0].statements[0]
+        assert "product_id UUID NOT NULL" in statement
+
+    def test_create_index_with_storing_clause(self):
+        """Integration test verifying DDL generation for indexes with STORING clause."""
+        add_result(
+            """SELECT true
+FROM INFORMATION_SCHEMA.TABLES
+WHERE TABLE_SCHEMA="" AND TABLE_NAME="items"
+LIMIT 1
+""",
+            ResultSet(),
+        )
+        engine = self.create_engine()
+        metadata = MetaData()
+        items = Table(
+            "items",
+            metadata,
+            Column("id", Uuid, primary_key=True),
+            Column("category", types.String(50)),
+            Column("name", types.String(100)),
+            Column("description", types.String(500)),
+        )
+        Index(
+            "ix_items_category",
+            items.c.category,
+            spanner_storing=["name", "description"],
+        )
+        metadata.create_all(engine)
+        requests = self.database_admin_service.requests
+        eq_(1, len(requests))
+        is_instance_of(requests[0], UpdateDatabaseDdlRequest)
+        statements = requests[0].statements
+        create_index_statement = [s for s in statements if "CREATE INDEX" in s][0]
+        assert "STORING (name, description)" in create_index_statement