Repository: GoogleCloudPlatform/cloud-sql-python-connector
Pull request #1040 (2024-03-18 18:31:13)

refactor: update defaulting of service endpoint
For TPC support we need to differentiate between if a user has passed in `sqladmin_api_endpoint` vs if its being defaulted. 

To do this we will default to `None` and use a global default endpoint `"https://sqladmin.googleapis.com"` when the endpoint is not explicitly set. This will allow for checking if a user has passed in a custom endpoint or not to support TPC cases.


Patch:
diff --git a/google/cloud/sql/connector/client.py b/google/cloud/sql/connector/client.py
index f28faa6..66a253b 100644
--- a/google/cloud/sql/connector/client.py
+++ b/google/cloud/sql/connector/client.py
@@ -31,6 +31,7 @@ if TYPE_CHECKING:
 
 USER_AGENT: str = f"cloud-sql-python-connector/{version}"
 API_VERSION: str = "v1beta4"
+DEFAULT_SERVICE_ENDPOINT: str = "https://sqladmin.googleapis.com"
 
 logger = logging.getLogger(name=__name__)
 
@@ -45,7 +46,7 @@ def _format_user_agent(driver: Optional[str], custom: Optional[str]) -> str:
 class CloudSQLClient:
     def __init__(
         self,
-        sqladmin_api_endpoint: str,
+        sqladmin_api_endpoint: Optional[str],
         quota_project: Optional[str],
         credentials: Credentials,
         client: Optional[aiohttp.ClientSession] = None,
@@ -81,7 +82,10 @@ class CloudSQLClient:
 
         self._client = client if client else aiohttp.ClientSession(headers=headers)
         self._credentials = credentials
-        self._sqladmin_api_endpoint = sqladmin_api_endpoint
+        if sqladmin_api_endpoint is None:
+            self._sqladmin_api_endpoint = DEFAULT_SERVICE_ENDPOINT
+        else:
+            self._sqladmin_api_endpoint = sqladmin_api_endpoint
         self._user_agent = user_agent
 
     async def _get_metadata(
diff --git a/google/cloud/sql/connector/connector.py b/google/cloud/sql/connector/connector.py
index f19e299..caba47e 100755
--- a/google/cloud/sql/connector/connector.py
+++ b/google/cloud/sql/connector/connector.py
@@ -93,7 +93,7 @@ class Connector:
         credentials: Optional[Credentials] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
         quota_project: Optional[str] = None,
-        sqladmin_api_endpoint: str = "https://sqladmin.googleapis.com",
+        sqladmin_api_endpoint: Optional[str] = None,
         user_agent: Optional[str] = None,
     ) -> None:
         # if event loop is given, use for background tasks
@@ -369,7 +369,7 @@ async def create_async_connector(
     credentials: Optional[Credentials] = None,
     loop: Optional[asyncio.AbstractEventLoop] = None,
     quota_project: Optional[str] = None,
-    sqladmin_api_endpoint: str = "https://sqladmin.googleapis.com",
+    sqladmin_api_endpoint: Optional[str] = None,
     user_agent: Optional[str] = None,
 ) -> Connector:
     """


Test patch:
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index da1c97e..1f5ce28 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -77,6 +77,21 @@ async def test_CloudSQLClient_init_(fake_credentials: FakeCredentials) -> None:
     await client.close()
 
 
+async def test_CloudSQLClient_init_default_service_endpoint(
+    fake_credentials: FakeCredentials,
+) -> None:
+    """
+    Test to check whether the __init__ method of CloudSQLClient
+    can correctly initialize the default service endpoint.
+    """
+    driver = "pg8000"
+    client = CloudSQLClient(None, "my-quota-project", fake_credentials, driver=driver)
+    # verify base endpoint is set to proper default
+    assert client._sqladmin_api_endpoint == "https://sqladmin.googleapis.com"
+    # close client
+    await client.close()
+
+
 @pytest.mark.asyncio
 async def test_CloudSQLClient_init_custom_user_agent(
     fake_credentials: FakeCredentials,