fix(sqlalchemy-spanner): include the constraint name in reflected primary keys

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

@@ -1435,7 +1435,7 @@ def get_multi_pk_constraint(
         table_type_query = self._get_table_type_query(kind, True)
 
         sql = """
-            SELECT tc.table_schema, tc.table_name, kcu.column_name
+            SELECT tc.table_schema, tc.table_name, kcu.column_name, tc.constraint_name
             FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
             JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS kcu
                 USING (TABLE_CATALOG, TABLE_SCHEMA, CONSTRAINT_NAME)
@@ -1459,11 +1459,10 @@ def get_multi_pk_constraint(
 
             for row in rows:
                 row[0] = row[0] or None
-                table_info = result_dict.get(
-                    (row[0], row[1]), {"constrained_columns": []}
-                )
-                table_info["constrained_columns"].append(row[2])
-                result_dict[(row[0], row[1])] = table_info
+                key = (row[0], row[1])
+                if key not in result_dict:
+                    result_dict[key] = {"name": row[3], "constrained_columns": []}
+                result_dict[key]["constrained_columns"].append(row[2])
 
         return result_dict
 

packages/sqlalchemy-spanner/tests/test_suite_20.py

@@ -891,6 +891,23 @@ def test_get_multi_pk_constraint(
             result = insp.get_multi_pk_constraint(**kw)
             self._check_table_dict(result, exp, self._required_pk_keys, make_lists=True)
 
+    def test_get_pk_constraint_name(self, connection):
+        """
+        SPANNER SPECIFIC TEST:
+
+        Spanner names primary key constraints PK_<table>. Checks that reflected
+        primary keys include that name, as ReflectedPrimaryKeyConstraint requires
+        and Alembic's autogenerate expects.
+        """
+        insp = inspect(connection)
+        eq_(
+            insp.get_pk_constraint("users"),
+            {"name": "PK_users", "constrained_columns": ["user_id"]},
+        )
+        result = insp.get_multi_pk_constraint(filter_names=["users", "dingalings"])
+        eq_(result[(None, "users")]["name"], "PK_users")
+        eq_(result[(None, "dingalings")]["name"], "PK_dingalings")
+
     def exp_fks(
         self,
         schema=None,

packages/sqlalchemy-spanner/tests/unit/test_dialect.py

@@ -37,6 +37,37 @@ def test_get_multi_indexes_excludes_search_indexes_sql(self):
         executed_sql = mock_snapshot.execute_sql.call_args[0][0]
         assert "i.index_type != 'SEARCH'" in executed_sql
 
+    def test_get_multi_pk_constraint_includes_name(self):
+        """Test reflected primary keys include the constraint name."""
+        dialect = SpannerDialect()
+        connection = MagicMock()
+        mock_snapshot = MagicMock()
+        # Mock rows: schema, table, column_name, constraint_name
+        mock_snapshot.execute_sql.return_value = [
+            ["", "singers", "singer_id", "PK_singers"],
+            ["", "albums", "singer_id", "PK_albums"],
+            ["", "albums", "album_id", "PK_albums"],
+        ]
+        connection.connection.database.snapshot.return_value.__enter__.return_value = (
+            mock_snapshot
+        )
+
+        eq_(
+            dialect.get_multi_pk_constraint(connection),
+            {
+                (None, "singers"): {
+                    "name": "PK_singers",
+                    "constrained_columns": ["singer_id"],
+                },
+                (None, "albums"): {
+                    "name": "PK_albums",
+                    "constrained_columns": ["singer_id", "album_id"],
+                },
+            },
+        )
+        executed_sql = mock_snapshot.execute_sql.call_args[0][0]
+        assert "tc.constraint_name" in executed_sql
+
     def test_get_multi_indexes_handles_none_column_ordering(self):
         """Test get_multi_indexes with None column ordering."""
         dialect = SpannerDialect()