fix(spanner): ensure session pool is bound before database use

Author: sinhasubham

google/cloud/spanner_v1/_async/database.py

@@ -14,10 +14,8 @@
 
 """User-friendly container for Cloud Spanner Database."""
 __CROSS_SYNC_OUTPUT__ = "google.cloud.spanner_v1.database"
-import asyncio
 import copy
 import functools
-import inspect
 import logging
 import re
 import threading
@@ -223,14 +221,8 @@ def __init__(
             pool = BurstyPool(database_role=database_role)
 
         self._pool = pool
-        res = pool.bind(self)
-        try:
-            loop = asyncio.get_running_loop()
-            if loop.is_running() and inspect.isawaitable(res):
-                loop.create_task(res)
-        except RuntimeError:
-            # No running loop, bind should have been sync or will be failed later
-            pass
+        # Note: self._pool.bind(self) should be called via Instance.database()
+        # factory method to ensure proper async initialization.
         self._experimental_host = (
             self._instance.experimental_host if self._instance else None
         )

google/cloud/spanner_v1/_async/instance.py

@@ -433,7 +433,8 @@ async def delete(self):
 
         await api.delete_instance(name=self.name, metadata=metadata)
 
-    def database(
+    @CrossSync.convert
+    async def database(
         self,
         database_id,
         ddl_statements=(),
@@ -498,7 +499,7 @@ def database(
         """
 
         if not enable_interceptors_in_tests:
-            return Database(
+            db = Database(
                 database_id,
                 self,
                 ddl_statements=ddl_statements,
@@ -511,7 +512,7 @@ def database(
                 proto_descriptors=proto_descriptors,
             )
         else:
-            return TestDatabase(
+            db = TestDatabase(
                 database_id,
                 self,
                 ddl_statements=ddl_statements,
@@ -523,6 +524,9 @@ def database(
                 enable_drop_protection=enable_drop_protection,
             )
 
+        await db._pool.bind(db)
+        return db
+
     @CrossSync.convert
     async def list_databases(self, page_size=None):
         """List databases for the instance.

google/cloud/spanner_v1/database.py

@@ -16,10 +16,8 @@
 # This file is automatically generated by CrossSync. Do not edit manually.
 
 """User-friendly container for Cloud Spanner Database."""
-import asyncio
 import copy
 import functools
-import inspect
 import logging
 import re
 import threading
@@ -203,13 +201,6 @@ def __init__(
         if pool is None:
             pool = BurstyPool(database_role=database_role)
         self._pool = pool
-        res = pool.bind(self)
-        try:
-            loop = asyncio.get_running_loop()
-            if loop.is_running() and inspect.isawaitable(res):
-                loop.create_task(res)
-        except RuntimeError:
-            pass
         self._experimental_host = (
             self._instance.experimental_host if self._instance else None
         )

google/cloud/spanner_v1/instance.py

@@ -455,7 +455,7 @@ def database(
         :rtype: :class:`~google.cloud.spanner_v1.database.Database`
         :returns: a database owned by this instance."""
         if not enable_interceptors_in_tests:
-            return Database(
+            db = Database(
                 database_id,
                 self,
                 ddl_statements=ddl_statements,
@@ -468,7 +468,7 @@ def database(
                 proto_descriptors=proto_descriptors,
             )
         else:
-            return TestDatabase(
+            db = TestDatabase(
                 database_id,
                 self,
                 ddl_statements=ddl_statements,
@@ -479,6 +479,8 @@ def database(
                 database_role=database_role,
                 enable_drop_protection=enable_drop_protection,
             )
+        db._pool.bind(db)
+        return db
 
     def list_databases(self, page_size=None):
         """List databases for the instance.

tests/_builders.py

@@ -174,6 +174,9 @@ def build_database(**kwargs: Mapping) -> Database:
     database = cls(**kwargs)
     database._spanner_api = build_spanner_api()
 
+    if not is_async:
+        database._pool.bind(database)
+
     return database
 
 

tests/system/_async/conftest.py

@@ -145,7 +145,7 @@ async def shared_database(
     pool = spanner_v1.AsyncBurstyPool(labels={"testcase": "database_api_async"})
 
     if database_dialect == DatabaseDialect.POSTGRESQL:
-        database = shared_instance.database(
+        database = await shared_instance.database(
             database_name,
             pool=pool,
             database_dialect=database_dialect,
@@ -156,7 +156,7 @@ async def shared_database(
         op = await database.update_ddl(ddl_statements=_helpers.DDL_STATEMENTS)
         await op.result(database_operation_timeout)
     else:
-        database = shared_instance.database(
+        database = await shared_instance.database(
             database_name,
             ddl_statements=_helpers.DDL_STATEMENTS,
             pool=pool,

tests/system/_async/test_database_api.py

@@ -36,7 +36,7 @@ async def test_table_not_found(shared_instance):
     )
     create_index = f"CREATE INDEX IDX ON {incorrect_table} (Field1)"
 
-    temp_db = shared_instance.database(
+    temp_db = await shared_instance.database(
         temp_db_id, ddl_statements=[create_table, create_index]
     )
     with pytest.raises(exceptions.NotFound):
@@ -55,7 +55,7 @@ async def test_list_databases(shared_instance, shared_database):
 async def test_create_database(shared_instance, databases_to_delete, database_dialect):
     pool = spanner_v1.AsyncBurstyPool(labels={"testcase": "create_database_async"})
     temp_db_id = _helpers.unique_id("temp_db_async")
-    temp_db = shared_instance.database(
+    temp_db = await shared_instance.database(
         temp_db_id, pool=pool, database_dialect=database_dialect
     )
     operation = await temp_db.create()