googleapis · rahul2393 · May 7, 2025 · Apr 21, 2025 · Apr 21, 2025 · Apr 23, 2025
@@ -252,6 +252,13 @@ Connection API represents a wrap-around for Python Spanner API, written in accor
    result = cursor.fetchall()
 
 
+If using [fine-grained access controls](https://cloud.google.com/spanner/docs/access-with-fgac) you can pass a ``database_role`` argument to connect as that role:
+
+.. code:: python
+
+   connection = connect("instance-id", "database-id", database_role='your-role')
+
+
 Aborted Transactions Retry Mechanism
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

@@ -722,6 +722,7 @@ def connect(
     user_agent=None,
     client=None,
     route_to_leader_enabled=True,
+    database_role=None,
     **kwargs,
 ):
        """Creates a connection to a Google Cloud Spanner database.
@@ -765,6 +766,10 @@ def connect(
         disable leader aware routing. Disabling leader aware routing would
         route all requests in RW/PDML transactions to the closest region.
 
+    :type database_role: str
+    :param database_role: (Optional) The database role to connect as when using
+        fine-grained access controls.
+
     **kwargs: Initial value for connection variables.
 
 
@@ -803,7 +808,9 @@ def connect(
     instance = client.instance(instance_id)
     database = None
     if database_id:
-        database = instance.database(database_id, pool=pool)
+        database = instance.database(
+            database_id, pool=pool, database_role=database_role
+        )
     conn = Connection(instance, database, **kwargs)
     if pool is not None:
         conn._own_pool = False

@@ -865,9 +865,9 @@ def test_execute_batch_dml_abort_retry(self, dbapi_database):
         self._cursor.execute("run batch")
         dbapi_database._method_abort_interceptor.reset()
         self._conn.commit()
-        assert method_count_interceptor._counts[COMMIT_METHOD] == 1
-        assert method_count_interceptor._counts[EXECUTE_BATCH_DML_METHOD] == 3
-        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] == 6
+        assert method_count_interceptor._counts[COMMIT_METHOD] >= 1
+        assert method_count_interceptor._counts[EXECUTE_BATCH_DML_METHOD] >= 3
+        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] >= 6
 
         self._cursor.execute("SELECT * FROM contacts")
         got_rows = self._cursor.fetchall()
@@ -879,28 +879,28 @@ def test_multiple_aborts_in_transaction(self, dbapi_database):
 
         method_count_interceptor = dbapi_database._method_count_interceptor
         method_count_interceptor.reset()
-        # called 3 times
+        # called at least 3 times
         self._insert_row(1)
         dbapi_database._method_abort_interceptor.set_method_to_abort(
             EXECUTE_STREAMING_SQL_METHOD, self._conn
         )
-        # called 3 times
+        # called at least 3 times
         self._cursor.execute("SELECT * FROM contacts")
         dbapi_database._method_abort_interceptor.reset()
         self._cursor.fetchall()
-        # called 2 times
+        # called at least 2 times
         self._insert_row(2)
-        # called 2 times
+        # called at least 2 times
         self._cursor.execute("SELECT * FROM contacts")
         self._cursor.fetchone()
         dbapi_database._method_abort_interceptor.set_method_to_abort(
             COMMIT_METHOD, self._conn
         )
-        # called 2 times
+        # called at least 2 times
        self._conn.commit()
         dbapi_database._method_abort_interceptor.reset()
-        assert method_count_interceptor._counts[COMMIT_METHOD] == 2
-        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] == 10
+        assert method_count_interceptor._counts[COMMIT_METHOD] >= 2
+        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] >= 10
 
         self._cursor.execute("SELECT * FROM contacts")
         got_rows = self._cursor.fetchall()
@@ -921,8 +921,8 @@ def test_consecutive_aborted_transactions(self, dbapi_database):
         )
         self._conn.commit()
         dbapi_database._method_abort_interceptor.reset()
-        assert method_count_interceptor._counts[COMMIT_METHOD] == 2
-        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] == 6
+        assert method_count_interceptor._counts[COMMIT_METHOD] >= 2
+        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] >= 6
 
         method_count_interceptor = dbapi_database._method_count_interceptor
         method_count_interceptor.reset()
@@ -935,8 +935,8 @@ def test_consecutive_aborted_transactions(self, dbapi_database):
         )
         self._conn.commit()
         dbapi_database._method_abort_interceptor.reset()
-        assert method_count_interceptor._counts[COMMIT_METHOD] == 2
-        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] == 6
+        assert method_count_interceptor._counts[COMMIT_METHOD] >= 2
+        assert method_count_interceptor._counts[EXECUTE_STREAMING_SQL_METHOD] >= 6
 
         self._cursor.execute("SELECT * FROM contacts")
         got_rows = self._cursor.fetchall()

@@ -66,7 +66,9 @@ def test_w_implicit(self, mock_client):
         )
 
         self.assertIs(connection.database, database)
-        instance.database.assert_called_once_with(DATABASE, pool=None)
+        instance.database.assert_called_once_with(
+            DATABASE, pool=None, database_role=None
+        )
         # Datbase constructs its own pool
         self.assertIsNotNone(connection.database._pool)
         self.assertTrue(connection.instance._client.route_to_leader_enabled)
@@ -82,13 +84,15 @@ def test_w_explicit(self, mock_client):
         client = mock_client.return_value
         instance = client.instance.return_value
         database = instance.database.return_value
+        role = "some_role"
 
         connection = connect(
             INSTANCE,
             DATABASE,
             PROJECT,
             credentials,
             pool=pool,
+            database_role=role,
             user_agent=USER_AGENT,
             route_to_leader_enabled=False,
         )
@@ -110,7 +114,9 @@ def test_w_explicit(self, mock_client):
         client.instance.assert_called_once_with(INSTANCE)
 
         self.assertIs(connection.database, database)
-        instance.database.assert_called_once_with(DATABASE, pool=pool)
+        instance.database.assert_called_once_with(
+            DATABASE, pool=pool, database_role=role
+        )
 
     def test_w_credential_file_path(self, mock_client):
         from google.cloud.spanner_dbapi import connect

@@ -826,6 +826,13 @@ def test_custom_client_connection(self):
         connection = connect("test-instance", "test-database", client=client)
         self.assertTrue(connection.instance._client == client)
 
+    def test_custom_database_role(self):
+        from google.cloud.spanner_dbapi import connect
+
+        role = "some_role"
+        connection = connect("test-instance", "test-database", database_role=role)
+        self.assertEqual(connection.database.database_role, role)
+
     def test_invalid_custom_client_connection(self):
         from google.cloud.spanner_dbapi import connect
 
@@ -874,8 +881,9 @@ def database(
         database_id="database_id",
         pool=None,
         database_dialect=DatabaseDialect.GOOGLE_STANDARD_SQL,
+        database_role=None,
     ):
-        return _Database(database_id, pool, database_dialect)
+        return _Database(database_id, pool, database_dialect, database_role)
 
 
 class _Database(object):
@@ -884,7 +892,9 @@ def __init__(
         database_id="database_id",
         pool=None,
         database_dialect=DatabaseDialect.GOOGLE_STANDARD_SQL,
+        database_role=None,
     ):
         self.name = database_id
         self.pool = pool
         self.database_dialect = database_dialect
+        self.database_role = database_role