Repository: GoogleCloudPlatform/cloud-sql-python-connector
Pull request #937 (2023-12-15 16:46:15)

feat: add support for domain-scoped projects
Domain-scoped project names (i.e. domain:project) were thought to be legacy but TPC universes will follow the pattern adding a TPC scoped prefix to projects. 

This PR adds support for instance connection names that include a domain-scoped project name.
(ex. google.com:my-project:my-region:my-instance) which would previously throw an error for invalid instance connection name.


Patch:
diff --git a/google/cloud/sql/connector/instance.py b/google/cloud/sql/connector/instance.py
index ebb9e08..7be5d07 100644
--- a/google/cloud/sql/connector/instance.py
+++ b/google/cloud/sql/connector/instance.py
@@ -18,6 +18,7 @@ from __future__ import annotations
 import asyncio
 from enum import Enum
 import logging
+import re
 import ssl
 from tempfile import TemporaryDirectory
 from typing import (
@@ -54,6 +55,22 @@ logger = logging.getLogger(name=__name__)
 
 APPLICATION_NAME = "cloud-sql-python-connector"
 
+# Instance connection name is the format <PROJECT>:<REGION>:<INSTANCE>
+# Additionally, we have to support legacy "domain-scoped" projects
+# (e.g. "google.com:PROJECT")
+CONN_NAME_REGEX = re.compile(("([^:]+(:[^:]+)?):([^:]+):([^:]+)"))
+
+
+def _parse_instance_connection_name(connection_name: str) -> Tuple[str, str, str]:
+    if CONN_NAME_REGEX.fullmatch(connection_name) is None:
+        raise ValueError(
+            "Arg `instance_connection_string` must have "
+            "format: PROJECT:REGION:INSTANCE, "
+            f"got {connection_name}."
+        )
+    connection_name_split = CONN_NAME_REGEX.split(connection_name)
+    return connection_name_split[1], connection_name_split[3], connection_name_split[4]
+
 
 class IPTypes(Enum):
     PUBLIC: str = "PRIMARY"
@@ -217,20 +234,11 @@ class Instance:
         quota_project: str = None,
         sqladmin_api_endpoint: str = "https://sqladmin.googleapis.com",
     ) -> None:
-        # Validate connection string
-        connection_string_split = instance_connection_string.split(":")
-
-        if len(connection_string_split) == 3:
-            self._instance_connection_string = instance_connection_string
-            self._project = connection_string_split[0]
-            self._region = connection_string_split[1]
-            self._instance = connection_string_split[2]
-        else:
-            raise ValueError(
-                "Arg `instance_connection_string` must have "
-                "format: PROJECT:REGION:INSTANCE, "
-                f"got {instance_connection_string}."
-            )
+        # validate and parse instance connection name
+        self._project, self._region, self._instance = _parse_instance_connection_name(
+            instance_connection_string
+        )
+        self._instance_connection_string = instance_connection_string
 
         self._enable_iam_auth = enable_iam_auth
 


Test patch:
diff --git a/tests/unit/test_instance.py b/tests/unit/test_instance.py
index 00fe59d..2c92d85 100644
--- a/tests/unit/test_instance.py
+++ b/tests/unit/test_instance.py
@@ -15,6 +15,7 @@ limitations under the License.
 """
 import asyncio
 import datetime
+from typing import Tuple
 
 from aiohttp import ClientResponseError, RequestInfo
 from aioresponses import aioresponses
@@ -29,6 +30,7 @@ from google.cloud.sql.connector.exceptions import (
     CredentialsTypeError,
 )
 from google.cloud.sql.connector.instance import (
+    _parse_instance_connection_name,
     ConnectionInfo,
     Instance,
     IPTypes,
@@ -42,6 +44,34 @@ def test_rate_limiter() -> AsyncRateLimiter:
     return AsyncRateLimiter(max_capacity=1, rate=1 / 2)
 
 
+@pytest.mark.parametrize(
+    "connection_name, expected",
+    [
+        ("project:region:instance", ("project", "region", "instance")),
+        (
+            "domain-prefix:project:region:instance",
+            ("domain-prefix:project", "region", "instance"),
+        ),
+    ],
+)
+def test_parse_instance_connection_name(
+    connection_name: str, expected: Tuple[str, str, str]
+) -> None:
+    """
+    Test that _parse_instance_connection_name works correctly on
+    normal instance connection names and domain-scoped projects.
+    """
+    assert expected == _parse_instance_connection_name(connection_name)
+
+
+def test_parse_instance_connection_name_bad_conn_name() -> None:
+    """
+    Tests that ValueError is thrown for bad instance connection names.
+    """
+    with pytest.raises(ValueError):
+        _parse_instance_connection_name("project:instance")  # missing region
+
+
 @pytest.mark.asyncio
 async def test_Instance_init(
     fake_credentials: Credentials,