From fa3efefc5719512216b9f32d0a7e7a9f1216c648 Mon Sep 17 00:00:00 2001 From: Warren Tian Date: Fri, 7 Aug 2026 16:31:01 -0700 Subject: [PATCH 1/5] feat: add universe domain support for TPC Add universe_domain parameter to Connector and AsyncConnector classes to support TPC (Trusted Partner Cloud) universe domains. Changes include: - Dynamic AlloyDB API endpoint construction based on universe domain - GOOGLE_CLOUD_UNIVERSE_DOMAIN env var fallback - Credential universe domain validation - Unit tests for all new functionality --- .../cloud/alloydbconnector/async_connector.py | 44 +++++++- google/cloud/alloydbconnector/connector.py | 42 ++++++- tests/unit/mocks.py | 16 +++ tests/unit/test_async_connector.py | 88 ++++++++++++++- tests/unit/test_connector.py | 104 +++++++++++++++++- 5 files changed, 284 insertions(+), 10 deletions(-) diff --git a/google/cloud/alloydbconnector/async_connector.py b/google/cloud/alloydbconnector/async_connector.py index d834f262..5458fdb9 100644 --- a/google/cloud/alloydbconnector/async_connector.py +++ b/google/cloud/alloydbconnector/async_connector.py @@ -16,6 +16,7 @@ import asyncio import logging +import os from types import TracebackType from typing import TYPE_CHECKING from typing import Any @@ -40,6 +41,10 @@ logger = logging.getLogger(name=__name__) +_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com" +_DEFAULT_ALLOYDB_API_ENDPOINT = "alloydb.googleapis.com" +_ALLOYDB_HOST_TEMPLATE = "alloydb.{universe_domain}" + class AsyncConnector: """A class to configure and create connections to Cloud SQL instances @@ -63,7 +68,10 @@ class AsyncConnector: billing purposes. Defaults to None, picking up project from environment. alloydb_api_endpoint (str): Base URL to use when calling - the AlloyDB API endpoint. Defaults to "alloydb.googleapis.com". + the AlloyDB API endpoint. Defaults to "alloydb.googleapis.com", + this argument should only be used in development. + universe_domain (str): The universe domain for AlloyDB API calls. + Default: "googleapis.com". enable_iam_auth (bool): Enables automatic IAM database authentication. ip_type (str | IPTypes): Default IP type for all AlloyDB connections. Defaults to IPTypes.PRIVATE ("PRIVATE") for private IP connections. @@ -84,11 +92,11 @@ def __init__( ip_type: str | IPTypes = IPTypes.PRIVATE, user_agent: Optional[str] = None, refresh_strategy: str | RefreshStrategy = RefreshStrategy.BACKGROUND, + universe_domain: Optional[str] = None, ) -> None: self._cache: dict[str, CacheTypes] = {} # initialize default params self._quota_project = quota_project - self._alloydb_api_endpoint = strip_http_prefix(alloydb_api_endpoint) self._enable_iam_auth = enable_iam_auth # if ip_type is str, convert to IPTypes enum if isinstance(ip_type, str): @@ -99,6 +107,22 @@ def __init__( refresh_strategy = RefreshStrategy(refresh_strategy.upper()) self._refresh_strategy = refresh_strategy self._user_agent = user_agent + # check for universe domain arg and then env var + if universe_domain: + self._universe_domain = universe_domain + else: + self._universe_domain = os.environ.get( + "GOOGLE_CLOUD_UNIVERSE_DOMAIN" + ) # type: ignore + # construct service endpoint for AlloyDB API calls + # if user has not overridden the endpoint, build it from universe domain + if alloydb_api_endpoint == _DEFAULT_ALLOYDB_API_ENDPOINT: + self._alloydb_api_endpoint = _ALLOYDB_HOST_TEMPLATE.format( + universe_domain=self.universe_domain + ) + else: + # user explicitly provided a custom endpoint, use it as-is + self._alloydb_api_endpoint = strip_http_prefix(alloydb_api_endpoint) # initialize credentials for authenticating with AlloyDB Admin API scopes = ["https://www.googleapis.com/auth/cloud-platform"] if credentials: @@ -106,6 +130,18 @@ def __init__( # otherwise use application default credentials else: self._credentials, _ = google.auth.default(scopes=scopes) + + # validate that the universe domain of the credentials matches the + # universe domain of the service endpoint + if self._credentials.universe_domain != self.universe_domain: + raise ValueError( + f"The configured universe domain ({self.universe_domain}) does " + "not match the universe domain found in the credentials " + f"({self._credentials.universe_domain}). If you haven't " + "configured the universe domain explicitly, `googleapis.com` " + "is the default." + ) + # initialize credentials for authenticating with the DB if db_credentials: self._db_credentials = db_credentials @@ -133,6 +169,10 @@ def __init__( self._client: Optional[AlloyDBClient] = None self._closed = False + @property + def universe_domain(self) -> str: + return self._universe_domain or _DEFAULT_UNIVERSE_DOMAIN + async def connect( self, instance_uri: str, diff --git a/google/cloud/alloydbconnector/connector.py b/google/cloud/alloydbconnector/connector.py index 3a269e91..a621e10c 100644 --- a/google/cloud/alloydbconnector/connector.py +++ b/google/cloud/alloydbconnector/connector.py @@ -20,6 +20,7 @@ from functools import partial import io import logging +import os import socket import struct from threading import Thread @@ -59,6 +60,10 @@ # the maximum amount of time to wait before aborting a metadata exchange IO_TIMEOUT = 30 +_DEFAULT_UNIVERSE_DOMAIN = "googleapis.com" +_DEFAULT_ALLOYDB_API_ENDPOINT = "alloydb.googleapis.com" +_ALLOYDB_HOST_TEMPLATE = "alloydb.{universe_domain}" + class Connector: """A class to configure and create connections to Cloud SQL instances. @@ -80,7 +85,10 @@ class Connector: billing purposes. Defaults to None, picking up project from environment. alloydb_api_endpoint (str): Base URL to use when calling - the AlloyDB API endpoint. Defaults to "alloydb.googleapis.com". + the AlloyDB API endpoint. Defaults to "alloydb.googleapis.com", + this argument should only be used in development. + universe_domain (str): The universe domain for AlloyDB API calls. + Default: "googleapis.com". enable_iam_auth (bool): Enables automatic IAM database authentication. ip_type (str | IPTypes): Default IP type for all AlloyDB connections. Defaults to IPTypes.PRIVATE ("PRIVATE") for private IP connections. @@ -108,6 +116,7 @@ def __init__( user_agent: Optional[str] = None, refresh_strategy: str | RefreshStrategy = RefreshStrategy.BACKGROUND, static_conn_info: Optional[io.TextIOBase] = None, + universe_domain: Optional[str] = None, ) -> None: # create event loop and start it in background thread self._loop: asyncio.AbstractEventLoop = asyncio.new_event_loop() @@ -116,7 +125,6 @@ def __init__( self._cache: dict[str, CacheTypes] = {} # initialize default params self._quota_project = quota_project - self._alloydb_api_endpoint = strip_http_prefix(alloydb_api_endpoint) self._enable_iam_auth = enable_iam_auth # if ip_type is str, convert to IPTypes enum if isinstance(ip_type, str): @@ -127,6 +135,20 @@ def __init__( refresh_strategy = RefreshStrategy(refresh_strategy.upper()) self._refresh_strategy = refresh_strategy self._user_agent = user_agent + # check for universe domain arg and then env var + if universe_domain: + self._universe_domain = universe_domain + else: + self._universe_domain = os.environ.get("GOOGLE_CLOUD_UNIVERSE_DOMAIN") + # construct service endpoint for AlloyDB API calls + # if user has not overridden the endpoint, build it from universe domain + if alloydb_api_endpoint == _DEFAULT_ALLOYDB_API_ENDPOINT: + self._alloydb_api_endpoint = _ALLOYDB_HOST_TEMPLATE.format( + universe_domain=self.universe_domain + ) + else: + # user explicitly provided a custom endpoint, use it as-is + self._alloydb_api_endpoint = strip_http_prefix(alloydb_api_endpoint) # initialize credentials for authenticating with AlloyDB Admin API scopes = ["https://www.googleapis.com/auth/cloud-platform"] if credentials: @@ -134,6 +156,18 @@ def __init__( # otherwise use application default credentials else: self._credentials, _ = default(scopes=scopes) + + # validate that the universe domain of the credentials matches the + # universe domain of the service endpoint + if self._credentials.universe_domain != self.universe_domain: + raise ValueError( + f"The configured universe domain ({self.universe_domain}) does " + "not match the universe domain found in the credentials " + f"({self._credentials.universe_domain}). If you haven't " + "configured the universe domain explicitly, `googleapis.com` " + "is the default." + ) + # initialize credentials for authenticating with the DB if db_credentials: self._db_credentials = db_credentials @@ -152,6 +186,10 @@ def __init__( self._static_conn_info = static_conn_info self._closed = False + @property + def universe_domain(self) -> str: + return self._universe_domain or _DEFAULT_UNIVERSE_DOMAIN + def connect(self, instance_uri: str, driver: str, **kwargs: Any) -> Any: """ Prepares and returns a database DBAPI connection object. diff --git a/tests/unit/mocks.py b/tests/unit/mocks.py index 9e6e174f..80de17d6 100644 --- a/tests/unit/mocks.py +++ b/tests/unit/mocks.py @@ -42,10 +42,14 @@ from google.cloud.alloydbconnector.connection_info import ConnectionInfo +from google.cloud.alloydbconnector.connector import _DEFAULT_UNIVERSE_DOMAIN + + class FakeCredentials: def __init__(self) -> None: self.token: Optional[str] = None self.expiry: Optional[datetime] = None + self._universe_domain = _DEFAULT_UNIVERSE_DOMAIN def refresh(self, _: Callable) -> None: """Refreshes the access token.""" @@ -67,6 +71,11 @@ def valid(self) -> bool: """Checks if the credentials are valid.""" return self.token is not None and not self.expired + @property + def universe_domain(self) -> str: + """The universe domain value.""" + return self._universe_domain + @property def token_state( self, @@ -98,6 +107,13 @@ def token_state( class FakeCredentialsRequiresScopes(Scoped): + _universe_domain = _DEFAULT_UNIVERSE_DOMAIN + + @property + def universe_domain(self) -> str: + """The universe domain value.""" + return self._universe_domain + def requires_scopes(self) -> bool: """ Overrides the requires_scopes() method of the Scoped class to require diff --git a/tests/unit/test_async_connector.py b/tests/unit/test_async_connector.py index 546b7a59..e583cb53 100644 --- a/tests/unit/test_async_connector.py +++ b/tests/unit/test_async_connector.py @@ -13,6 +13,7 @@ # limitations under the License. import asyncio +import os from typing import Any from typing import Union @@ -440,9 +441,9 @@ async def test_Connector_remove_cached_bad_instance( transport._wrapped_methods[transport.get_connection_info]._retry = AsyncRetry( timeout=1 ) - transport._wrapped_methods[ - transport.generate_client_certificate - ]._retry = AsyncRetry(timeout=1) + transport._wrapped_methods[transport.generate_client_certificate]._retry = ( + AsyncRetry(timeout=1) + ) with pytest.raises(RetryError): await connector.connect(instance_uri, "asyncpg") @@ -491,3 +492,84 @@ async def test_connect_when_closed(credentials: FakeCredentials) -> None: exc_info.value.args[0] == "Connection attempt failed because the connector has already been closed." ) + + +def test_async_default_universe_domain() -> None: + """Test that default universe domain and constructed service endpoint are + formatted correctly. + """ + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + connector = AsyncConnector(credentials=credentials) + # test universe domain was not configured + assert connector._universe_domain is None + # test property and service endpoint construction + assert connector.universe_domain == "googleapis.com" + assert connector._alloydb_api_endpoint == "alloydb.googleapis.com" + + +def test_async_configured_universe_domain_matches_GDU() -> None: + """Test that configured universe domain succeeds with matched GDU credentials.""" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + universe_domain = "googleapis.com" + connector = AsyncConnector(credentials=credentials, universe_domain=universe_domain) + assert connector._universe_domain == universe_domain + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + + +def test_async_configured_universe_domain_matches_credentials() -> None: + """Test that configured universe domain succeeds with matching universe + domain credentials. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + credentials._universe_domain = universe_domain + connector = AsyncConnector(credentials=credentials, universe_domain=universe_domain) + assert connector._universe_domain == universe_domain + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + + +def test_async_configured_universe_domain_mismatched_credentials() -> None: + """Test that configured universe domain errors with mismatched universe + domain credentials. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + with pytest.raises(ValueError) as exc_info: + AsyncConnector(credentials=credentials, universe_domain=universe_domain) + err_msg = ( + f"The configured universe domain ({universe_domain}) does " + "not match the universe domain found in the credentials " + f"({credentials.universe_domain}). If you haven't " + "configured the universe domain explicitly, `googleapis.com` " + "is the default." + ) + assert exc_info.value.args[0] == err_msg + + +def test_async_configured_universe_domain_env_var() -> None: + """Test that configured universe domain succeeds with universe + domain set via GOOGLE_CLOUD_UNIVERSE_DOMAIN env var. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + credentials._universe_domain = universe_domain + os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] = universe_domain + try: + connector = AsyncConnector(credentials=credentials) + assert connector._universe_domain == universe_domain + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + finally: + del os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] diff --git a/tests/unit/test_connector.py b/tests/unit/test_connector.py index c5d366eb..380ad2aa 100644 --- a/tests/unit/test_connector.py +++ b/tests/unit/test_connector.py @@ -13,6 +13,7 @@ # limitations under the License. import asyncio +import os from threading import Thread from typing import Union @@ -346,9 +347,9 @@ def test_Connector_remove_cached_bad_instance( transport._wrapped_methods[transport.get_connection_info]._retry = Retry( timeout=1 ) - transport._wrapped_methods[ - transport.generate_client_certificate - ]._retry = Retry(timeout=1) + transport._wrapped_methods[transport.generate_client_certificate]._retry = ( + Retry(timeout=1) + ) with pytest.raises(RetryError): connector.connect(instance_uri, "pg8000") @@ -420,3 +421,100 @@ def test_connect_when_closed(credentials: FakeCredentials) -> None: exc_info.value.args[0] == "Connection attempt failed because the connector has already been closed." ) + + +def test_default_universe_domain() -> None: + """Test that default universe domain and constructed service endpoint are + formatted correctly. + """ + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + with Connector(credentials=credentials) as connector: + # test universe domain was not configured + assert connector._universe_domain is None + # test property and service endpoint construction + assert connector.universe_domain == "googleapis.com" + assert connector._alloydb_api_endpoint == "alloydb.googleapis.com" + + +def test_configured_universe_domain_matches_GDU() -> None: + """Test that configured universe domain succeeds with matched GDU credentials.""" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + universe_domain = "googleapis.com" + with Connector( + credentials=credentials, universe_domain=universe_domain + ) as connector: + # test universe domain was configured + assert connector._universe_domain == universe_domain + # test property and service endpoint construction + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + + +def test_configured_universe_domain_matches_credentials() -> None: + """Test that configured universe domain succeeds with matching universe + domain credentials. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + # set fake credentials to be configured for the universe domain + credentials._universe_domain = universe_domain + with Connector( + credentials=credentials, universe_domain=universe_domain + ) as connector: + # test universe domain was configured + assert connector._universe_domain == universe_domain + # test property and service endpoint construction + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + + +def test_configured_universe_domain_mismatched_credentials() -> None: + """Test that configured universe domain errors with mismatched universe + domain credentials. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + # credentials have GDU domain ("googleapis.com") + with pytest.raises(ValueError) as exc_info: + Connector(credentials=credentials, universe_domain=universe_domain) + err_msg = ( + f"The configured universe domain ({universe_domain}) does " + "not match the universe domain found in the credentials " + f"({credentials.universe_domain}). If you haven't " + "configured the universe domain explicitly, `googleapis.com` " + "is the default." + ) + assert exc_info.value.args[0] == err_msg + + +def test_configured_universe_domain_env_var() -> None: + """Test that configured universe domain succeeds with universe + domain set via GOOGLE_CLOUD_UNIVERSE_DOMAIN env var. + """ + universe_domain = "test-universe.test" + credentials = FakeCredentials() + credentials.token = "test-token" + credentials.expiry = None + # set fake credentials to be configured for the universe domain + credentials._universe_domain = universe_domain + # set environment variable + os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] = universe_domain + # Note: we are not passing universe_domain arg, env var should set it + try: + with Connector(credentials=credentials) as connector: + # test universe domain was configured + assert connector._universe_domain == universe_domain + # test property and service endpoint construction + assert connector.universe_domain == universe_domain + assert connector._alloydb_api_endpoint == f"alloydb.{universe_domain}" + finally: + # unset env var + del os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] From f74ecd50036afef7fd424e8804441dc06daad102 Mon Sep 17 00:00:00 2001 From: Warren Tian Date: Fri, 7 Aug 2026 16:47:01 -0700 Subject: [PATCH 2/5] format --- google/cloud/alloydbconnector/async_connector.py | 4 +--- tests/unit/test_async_connector.py | 6 +++--- tests/unit/test_connector.py | 6 +++--- 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/google/cloud/alloydbconnector/async_connector.py b/google/cloud/alloydbconnector/async_connector.py index 5458fdb9..d3639c62 100644 --- a/google/cloud/alloydbconnector/async_connector.py +++ b/google/cloud/alloydbconnector/async_connector.py @@ -111,9 +111,7 @@ def __init__( if universe_domain: self._universe_domain = universe_domain else: - self._universe_domain = os.environ.get( - "GOOGLE_CLOUD_UNIVERSE_DOMAIN" - ) # type: ignore + self._universe_domain = os.environ.get("GOOGLE_CLOUD_UNIVERSE_DOMAIN") # type: ignore # construct service endpoint for AlloyDB API calls # if user has not overridden the endpoint, build it from universe domain if alloydb_api_endpoint == _DEFAULT_ALLOYDB_API_ENDPOINT: diff --git a/tests/unit/test_async_connector.py b/tests/unit/test_async_connector.py index e583cb53..1baf4471 100644 --- a/tests/unit/test_async_connector.py +++ b/tests/unit/test_async_connector.py @@ -441,9 +441,9 @@ async def test_Connector_remove_cached_bad_instance( transport._wrapped_methods[transport.get_connection_info]._retry = AsyncRetry( timeout=1 ) - transport._wrapped_methods[transport.generate_client_certificate]._retry = ( - AsyncRetry(timeout=1) - ) + transport._wrapped_methods[ + transport.generate_client_certificate + ]._retry = AsyncRetry(timeout=1) with pytest.raises(RetryError): await connector.connect(instance_uri, "asyncpg") diff --git a/tests/unit/test_connector.py b/tests/unit/test_connector.py index 380ad2aa..10da5f0b 100644 --- a/tests/unit/test_connector.py +++ b/tests/unit/test_connector.py @@ -344,9 +344,9 @@ def test_Connector_remove_cached_bad_instance( "alloydb.googleapis.com", "test-project", credentials, driver="pg8000" ) transport = connector._client._client.transport - transport._wrapped_methods[transport.get_connection_info]._retry = Retry( - timeout=1 - ) + transport._wrapped_methods[ + transport.generate_client_certificate + ]._retry = Retry(timeout=1) transport._wrapped_methods[transport.generate_client_certificate]._retry = ( Retry(timeout=1) ) From d562278faa9ba64d7f57f6b15eaac4b513bc0065 Mon Sep 17 00:00:00 2001 From: Warren Tian Date: Fri, 7 Aug 2026 16:48:29 -0700 Subject: [PATCH 3/5] format 2 --- tests/unit/test_connector.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_connector.py b/tests/unit/test_connector.py index 10da5f0b..17a5c493 100644 --- a/tests/unit/test_connector.py +++ b/tests/unit/test_connector.py @@ -347,9 +347,9 @@ def test_Connector_remove_cached_bad_instance( transport._wrapped_methods[ transport.generate_client_certificate ]._retry = Retry(timeout=1) - transport._wrapped_methods[transport.generate_client_certificate]._retry = ( - Retry(timeout=1) - ) + transport._wrapped_methods[ + transport.generate_client_certificate + ]._retry = Retry(timeout=1) with pytest.raises(RetryError): connector.connect(instance_uri, "pg8000") From aeba2b2d3f447382e11cfcc6baf72b63cd2aae02 Mon Sep 17 00:00:00 2001 From: Warren Tian Date: Fri, 7 Aug 2026 16:50:03 -0700 Subject: [PATCH 4/5] style: sort imports in mocks.py --- tests/unit/mocks.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/unit/mocks.py b/tests/unit/mocks.py index 80de17d6..8d350171 100644 --- a/tests/unit/mocks.py +++ b/tests/unit/mocks.py @@ -40,8 +40,6 @@ from google.cloud import alloydb_v1beta import google.cloud.alloydb_connectors_v1.proto.resources_pb2 as connectorspb from google.cloud.alloydbconnector.connection_info import ConnectionInfo - - from google.cloud.alloydbconnector.connector import _DEFAULT_UNIVERSE_DOMAIN From 79a5d970dfa75b08cf04f3b109c804c51d15d1af Mon Sep 17 00:00:00 2001 From: Warren Tian Date: Fri, 7 Aug 2026 16:53:51 -0700 Subject: [PATCH 5/5] fix: add Optional[str] type annotation for _universe_domain --- google/cloud/alloydbconnector/async_connector.py | 4 ++-- google/cloud/alloydbconnector/connector.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/google/cloud/alloydbconnector/async_connector.py b/google/cloud/alloydbconnector/async_connector.py index d3639c62..11ce61e2 100644 --- a/google/cloud/alloydbconnector/async_connector.py +++ b/google/cloud/alloydbconnector/async_connector.py @@ -109,9 +109,9 @@ def __init__( self._user_agent = user_agent # check for universe domain arg and then env var if universe_domain: - self._universe_domain = universe_domain + self._universe_domain: Optional[str] = universe_domain else: - self._universe_domain = os.environ.get("GOOGLE_CLOUD_UNIVERSE_DOMAIN") # type: ignore + self._universe_domain = os.environ.get("GOOGLE_CLOUD_UNIVERSE_DOMAIN") # construct service endpoint for AlloyDB API calls # if user has not overridden the endpoint, build it from universe domain if alloydb_api_endpoint == _DEFAULT_ALLOYDB_API_ENDPOINT: diff --git a/google/cloud/alloydbconnector/connector.py b/google/cloud/alloydbconnector/connector.py index a621e10c..dcf0dddd 100644 --- a/google/cloud/alloydbconnector/connector.py +++ b/google/cloud/alloydbconnector/connector.py @@ -137,7 +137,7 @@ def __init__( self._user_agent = user_agent # check for universe domain arg and then env var if universe_domain: - self._universe_domain = universe_domain + self._universe_domain: Optional[str] = universe_domain else: self._universe_domain = os.environ.get("GOOGLE_CLOUD_UNIVERSE_DOMAIN") # construct service endpoint for AlloyDB API calls