feat: support database role in connect arguments

Author: olavloite

samples/database_role.py

@@ -0,0 +1,42 @@
+# 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 uuid
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from sample_helper import run_sample
+from model import Singer
+
+
+# Shows how to set the database role for a connection.
+def database_role_sample():
+    engine = create_engine(
+        "spanner:///projects/sample-project/"
+        "instances/sample-instance/"
+        "databases/sample-database",
+        # You can set the database role in the connect arguments.
+        connect_args={"database_role": "my_role"},
+        echo=True,
+    )
+    with Session(engine) as session:
+        singer_id = str(uuid.uuid4())
+        singer = Singer(id=singer_id, first_name="John", last_name="Doe")
+        session.add(singer)
+        session.commit()
+
+
+if __name__ == "__main__":
+    run_sample(database_role_sample)

test/mockserver_tests/test_basics.py

@@ -267,6 +267,25 @@ class Singer(Base):
             session.add(singer)
             session.commit()
 
+    def test_database_role(self):
+        add_select1_result()
+        engine = create_engine(
+            "spanner:///projects/p/instances/i/databases/d",
+            connect_args={
+                "client": self.client,
+                "pool": FixedSizePool(size=10),
+                "database_role": "my_role",
+            },
+        )
+        with Session(engine.execution_options(isolation_level="autocommit")) as session:
+            session.execute(select(1))
+        requests = self.spanner_service.requests
+        eq_(2, len(requests))
+        is_instance_of(requests[0], BatchCreateSessionsRequest)
+        is_instance_of(requests[1], ExecuteSqlRequest)
+        request: BatchCreateSessionsRequest = requests[0]
+        eq_("my_role", request.session_template.creator_role)
+
     def test_select_table_in_named_schema(self):
         class Base(DeclarativeBase):
             pass