chore: add check for closed connector (#1269)

Author: kgala2

google/cloud/sql/connector/connector.py

@@ -34,6 +34,7 @@
 from google.cloud.sql.connector.enums import DriverMapping
 from google.cloud.sql.connector.enums import IPTypes
 from google.cloud.sql.connector.enums import RefreshStrategy
+from google.cloud.sql.connector.exceptions import ClosedConnectorError
 from google.cloud.sql.connector.exceptions import ConnectorLoopError
 from google.cloud.sql.connector.instance import RefreshAheadCache
 from google.cloud.sql.connector.lazy import LazyRefreshCache
@@ -155,6 +156,7 @@ def __init__(
         # connection name string and enable_iam_auth boolean flag
         self._cache: dict[tuple[str, bool], MonitoredCache] = {}
         self._client: Optional[CloudSQLClient] = None
+        self._closed: bool = False
 
         # initialize credentials
         scopes = ["https://www.googleapis.com/auth/sqlservice.admin"]
@@ -244,6 +246,12 @@ def connect(
         # connect runs sync database connections on background thread.
         # Async database connections should call 'connect_async' directly to
         # avoid hanging indefinitely.
+
+        # Check if the connector is closed before attempting to connect.
+        if self._closed:
+            raise ClosedConnectorError(
+                "Connection attempt failed because the connector has already been closed."
+            )
         connect_future = asyncio.run_coroutine_threadsafe(
             self.connect_async(instance_connection_string, driver, **kwargs),
             self._loop,
@@ -281,7 +289,13 @@ async def connect_async(
                 and then subsequent attempt with IAM database authentication.
             KeyError: Unsupported database driver Must be one of pymysql, asyncpg,
                 pg8000, and pytds.
+            RuntimeError: Connector has been closed. Cannot connect using a closed
+                Connector.
         """
+        if self._closed:
+            raise ClosedConnectorError(
+                "Connection attempt failed because the connector has already been closed."
+            )
         # check if event loop is running in current thread
         if self._loop != asyncio.get_running_loop():
             raise ConnectorLoopError(
@@ -477,9 +491,10 @@ def close(self) -> None:
     async def close_async(self) -> None:
         """Helper function to cancel the cache's tasks
         and close aiohttp.ClientSession."""
-        await asyncio.gather(*[cache.close() for cache in self._cache.values()])
+        self._closed = True
         if self._client:
             await self._client.close()
+        await asyncio.gather(*[cache.close() for cache in self._cache.values()])
 
 
 async def create_async_connector(

google/cloud/sql/connector/exceptions.py

@@ -84,3 +84,10 @@ class CacheClosedError(Exception):
     Exception to be raised when a ConnectionInfoCache can not be accessed after
     it is closed.
     """
+
+
+class ClosedConnectorError(Exception):
+    """
+    Exception to be raised when a Connector is closed and connect method is
+    called on it.
+    """

tests/unit/test_connector.py

@@ -29,6 +29,7 @@
 from google.cloud.sql.connector import IPTypes
 from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.connection_name import ConnectionName
+from google.cloud.sql.connector.exceptions import ClosedConnectorError
 from google.cloud.sql.connector.exceptions import CloudSQLIPTypeError
 from google.cloud.sql.connector.exceptions import ConnectorLoopError
 from google.cloud.sql.connector.exceptions import IncompatibleDriverError
@@ -502,3 +503,48 @@ def test_configured_quota_project_env_var(
         assert connector._quota_project == quota_project
     # unset env var
     del os.environ["GOOGLE_CLOUD_QUOTA_PROJECT"]
+
+
+@pytest.mark.asyncio
+async def test_connect_async_closed_connector(
+    fake_credentials: Credentials, fake_client: CloudSQLClient
+) -> None:
+    """Test that calling connect_async() on a closed connector raises an error."""
+    async with Connector(
+        credentials=fake_credentials, loop=asyncio.get_running_loop()
+    ) as connector:
+        connector._client = fake_client
+        await connector.close_async()
+        with pytest.raises(ClosedConnectorError) as exc_info:
+            await connector.connect_async(
+                "test-project:test-region:test-instance",
+                "asyncpg",
+                user="my-user",
+                password="my-pass",
+                db="my-db",
+            )
+        assert (
+            exc_info.value.args[0]
+            == "Connection attempt failed because the connector has already been closed."
+        )
+
+
+def test_connect_closed_connector(
+    fake_credentials: Credentials, fake_client: CloudSQLClient
+) -> None:
+    """Test that calling connect() on a closed connector raises an error."""
+    with Connector(credentials=fake_credentials) as connector:
+        connector._client = fake_client
+        connector.close()
+        with pytest.raises(ClosedConnectorError) as exc_info:
+            connector.connect(
+                "test-project:test-region:test-instance",
+                "pg8000",
+                user="my-user",
+                password="my-pass",
+                db="my-db",
+            )
+        assert (
+            exc_info.value.args[0]
+            == "Connection attempt failed because the connector has already been closed."
+        )