feat: support commit timestamp option

README.rst

@@ -234,6 +234,22 @@ tables with this feature, make sure to call ``add_is_dependent_on()`` on
 the child table to request SQLAlchemy to create the parent table before
 the child table.
 
+Commit timestamps
+~~~~~~~~~~~~~~~~~~
+
+The dialect offers the ``spanner_allow_commit_timestamp`` option to
+column constructors for creating commit timestamp columns.
+
+.. code:: python
+
+   Table(
+       "table",
+       metadata,
+       Column("last_update_time", DateTime, spanner_allow_commit_timestamp=True),
+   )
+
+`See this documentation page for more details <https://cloud.google.com/spanner/docs/commit-timestamp>`__.
+
 Unique constraints
 ~~~~~~~~~~~~~~~~~~
 

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -578,6 +578,11 @@ def get_column_specification(self, column, **kwargs):
         elif hasattr(column, "computed") and column.computed is not None:
             colspec += " " + self.process(column.computed)
 
+        if column.dialect_options.get("spanner", {}).get(
+            "allow_commit_timestamp", False
+        ):
+            colspec += " OPTIONS (allow_commit_timestamp=true)"
+
         return colspec
 
     def visit_computed_column(self, generated, **kw):

samples/model.py

@@ -33,6 +33,7 @@
     TextClause,
     Index,
     PickleType,
+    text,
 )
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import SpannerPickleType
@@ -75,6 +76,21 @@ class Singer(Base):
     concerts: Mapped[List["Concert"]] = relationship(
         back_populates="singer", cascade="all, delete-orphan"
     )
+    # Create a commit timestamp column and set a client-side default of
+    # PENDING_COMMIT_TIMESTAMP() An event handler below is responsible for
+    # setting PENDING_COMMIT_TIMESTAMP() on updates. If using SQLAlchemy
+    # core rather than the ORM, callers will need to supply their own
+    # PENDING_COMMIT_TIMESTAMP() values in their inserts & updates.
+    last_update_time: Mapped[datetime.datetime] = mapped_column(
+        spanner_allow_commit_timestamp=True,
+        default=text("PENDING_COMMIT_TIMESTAMP()"),
+    )
+
+
+@event.listens_for(TimestampUser, "before_update")
+def singer_before_update(mapper, connection, target):
+    """Updates the commit timestamp when the row is updated."""
+    target.updated_at = text("PENDING_COMMIT_TIMESTAMP()")
 
 
 class Album(Base):

test/mockserver_tests/commit_timestamp_model.py

@@ -0,0 +1,32 @@
+# Copyright 2025 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.
+
+import datetime
+
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class Singer(Base):
+    __tablename__ = "singers"
+    id: Mapped[str] = mapped_column(primary_key=True)
+    name: Mapped[str]
+    updated_at: Mapped[datetime.datetime] = mapped_column(
+        spanner_allow_commit_timestamp=True
+    )

test/mockserver_tests/test_commit_timestamp.py

@@ -0,0 +1,64 @@
+# Copyright 2025 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 sqlalchemy import create_engine
+from sqlalchemy.testing import eq_, is_instance_of
+from google.cloud.spanner_v1 import (
+    FixedSizePool,
+    ResultSet,
+)
+from test.mockserver_tests.mock_server_test_base import (
+    MockServerTestBase,
+    add_result,
+)
+from google.cloud.spanner_admin_database_v1 import UpdateDatabaseDdlRequest
+
+
+class TestCommitTimestamp(MockServerTestBase):
+    def test_create_table(self):
+        from test.mockserver_tests.commit_timestamp_model import Base
+
+        add_result(
+            """SELECT true
+FROM INFORMATION_SCHEMA.TABLES
+WHERE TABLE_SCHEMA="" AND TABLE_NAME="singers"
+LIMIT 1
+""",
+            ResultSet(),
+        )
+        add_result(
+            """SELECT true
+                FROM INFORMATION_SCHEMA.SEQUENCES
+                WHERE NAME="singer_id"
+                AND SCHEMA=""
+                LIMIT 1""",
+            ResultSet(),
+        )
+        engine = create_engine(
+            "spanner:///projects/p/instances/i/databases/d",
+            connect_args={"client": self.client, "pool": FixedSizePool(size=10)},
+        )
+        Base.metadata.create_all(engine)
+        requests = self.database_admin_service.requests
+        eq_(1, len(requests))
+        is_instance_of(requests[0], UpdateDatabaseDdlRequest)
+        eq_(1, len(requests[0].statements))
+        eq_(
+            "CREATE TABLE singers (\n"
+            "\tid STRING(MAX) NOT NULL, \n"
+            "\tname STRING(MAX) NOT NULL, \n"
+            "\tupdated_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)\n"
+            ") PRIMARY KEY (id)",
+            requests[0].statements[0],
+        )

test/system/test_basics.py

@@ -11,6 +11,7 @@
 # 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.
+import datetime
 import os
 from typing import Optional
 from sqlalchemy import (
@@ -29,10 +30,11 @@
     select,
     update,
     delete,
+    event,
 )
 from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column
 from sqlalchemy.types import REAL
-from sqlalchemy.testing import eq_, is_true
+from sqlalchemy.testing import eq_, is_true, is_not_none
 from sqlalchemy.testing.plugin.plugin_base import fixtures
 
 
@@ -264,6 +266,60 @@ class User(Base):
         eq_(len(inserted_rows), len(selected_rows))
         eq_(set(inserted_rows), set(selected_rows))
 
+    def test_commit_timestamp(self, connection):
+        """Ensures commit timestamps are set."""
+
+        class Base(DeclarativeBase):
+            pass
+
+        class TimestampUser(Base):
+            __tablename__ = "timestamp_users"
+            ID: Mapped[int] = mapped_column(primary_key=True)
+            name: Mapped[str]
+            updated_at: Mapped[datetime.datetime] = mapped_column(
+                spanner_allow_commit_timestamp=True,
+                default=text("PENDING_COMMIT_TIMESTAMP()"),
+            )
+
+        @event.listens_for(TimestampUser, "before_update")
+        def before_update(mapper, connection, target):
+            target.updated_at = text("PENDING_COMMIT_TIMESTAMP()")
+
+        engine = connection.engine
+        Base.metadata.create_all(engine)
+        try:
+            with Session(engine) as session:
+                session.add(TimestampUser(name="name"))
+                session.commit()
+
+            with Session(engine) as session:
+                users = list(
+                    session.scalars(
+                        select(TimestampUser).where(TimestampUser.name == "name")
+                    )
+                )
+                user = users[0]
+
+                is_not_none(user.updated_at)
+                created_at = user.updated_at
+
+                user.name = "new-name"
+                session.commit()
+
+            with Session(engine) as session:
+                users = list(
+                    session.scalars(
+                        select(TimestampUser).where(TimestampUser.name == "new-name")
+                    )
+                )
+                user = users[0]
+
+                is_not_none(user.updated_at)
+                is_true(user.updated_at > created_at)
+
+        finally:
+            Base.metadata.drop_all(engine)
+
     def test_cross_schema_fk_lookups(self, connection):
         """Ensures we introspect FKs within & across schema."""