feat(generator): gapic generator centralization routing

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -12,13 +12,157 @@ Add conditional logic to check if static code exists in google-api-core and use
 falling back to the local implementation if not present. #}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17883): 
 Backfill compatibility functions being removed from the client layer. #}
+"""A compatibility module for older versions of google-api-core."""
 
-{% if has_auto_populated_fields %}
-from typing import Union
 import uuid
 
+from typing import Optional, Union
+from urllib.parse import urlparse, urlunparse
+
+from google.auth.exceptions import MutualTLSChannelError
+from google.api_core.universe import EmptyUniverseError
 import google.protobuf.message
 
+DEFAULT_UNIVERSE = "googleapis.com"
+
+
+def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+    """Converts api endpoint to mTLS endpoint.
+
+    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+    Other URLs (including those that do not match these domain suffixes or
+    already contain '.mtls.') are passed through as-is.
+
+    Args:
+        api_endpoint (Optional[str]): the api endpoint to convert.
+
+    Returns:
+        Optional[str]: converted mTLS api endpoint.
+    """
+    if not api_endpoint or ".mtls." in api_endpoint.lower():
+        return api_endpoint
+
+    has_scheme = "://" in api_endpoint
+    if not has_scheme:
+        parsed = urlparse("//" + api_endpoint)
+    else:
+        parsed = urlparse(api_endpoint)
+
+    host = parsed.hostname
+    if not host:
+        return api_endpoint
+
+    port = f":{parsed.port}" if parsed.port else ""
+
+    lowered_host = host.lower()
+    suffix_sandbox = ".sandbox.googleapis.com"
+    suffix_google = ".googleapis.com"
+    if lowered_host.endswith(suffix_sandbox):
+        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+    elif lowered_host.endswith(suffix_google):
+        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+    else:
+        return api_endpoint
+
+    netloc = new_host + port
+    new_parsed = parsed._replace(netloc=netloc)
+
+    if not has_scheme:
+        return urlunparse(new_parsed)[2:]
+    else:
+        return urlunparse(new_parsed)
+
+def get_api_endpoint(
+    api_override: Optional[str],
+    universe_domain: str,
+    default_universe: str,
+    default_mtls_endpoint: Optional[str],
+    default_endpoint_template: str,
+    use_mtls: bool,
+) -> str:
+    """Return the API endpoint used by the client.
+
+    Args:
+        api_override (Optional[str]): The API endpoint override. If specified,
+            this is always returned.
+        universe_domain (str): The universe domain used by the client.
+        default_universe (str): The default universe domain.
+        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+        default_endpoint_template (str): The default endpoint template containing
+            a placeholder `{UNIVERSE_DOMAIN}`.
+        use_mtls (bool): Whether to use the mTLS endpoint.
+
+    Returns:
+        str: The API endpoint to be used by the client.
+
+    Raises:
+        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+            not supported in the configured universe domain.
+        ValueError: If mTLS is requested but no mTLS endpoint is available.
+    """
+    if api_override is not None:
+        return api_override
+
+    if use_mtls:
+        if universe_domain.lower() != default_universe.lower():
+            raise MutualTLSChannelError(
+                f"mTLS is not supported in any universe other than {default_universe}."
+            )
+        if not default_mtls_endpoint:
+            raise ValueError("mTLS endpoint is not available.")
+        return default_mtls_endpoint
+    else:
+        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+def get_universe_domain(
+    *potential_universes: Optional[str],
+    default_universe: str,
+) -> str:
+    """Return the universe domain used by the client.
+
+    Args:
+        *potential_universes (Optional[str]): Potential universe domains in order of preference.
+        default_universe (str): The default universe domain.
+
+    Returns:
+        str: The universe domain to be used by the client.
+
+    Raises:
+        EmptyUniverseError: If the resolved universe domain is an empty string.
+    """
+    resolved = next(
+        (x.strip() for x in potential_universes if x is not None),
+        default_universe,
+    )
+
+    if not resolved:
+        raise EmptyUniverseError()
+    return resolved
+
+def determine_domain(
+    client_universe_domain: Optional[str], universe_domain_env: Optional[str]
+) -> str:
+    """Return the universe domain used by the client.
+
+    Args:
+        client_universe_domain (Optional[str]): The universe domain configured via the client options.
+        universe_domain_env (Optional[str]): The universe domain configured via the
+        "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable.
+
+    Returns:
+        str: The universe domain to be used by the client.
+
+    Raises:
+        ValueError: If the universe domain is an empty string.
+    """
+    return get_universe_domain(
+        client_universe_domain,
+        universe_domain_env,
+        default_universe=DEFAULT_UNIVERSE,
+    )
+
+{% if has_auto_populated_fields %}
 
 def setup_request_id(
     request: Union[google.protobuf.message.Message, dict, None],
@@ -63,5 +207,6 @@ def setup_request_id(
     else:
         if not getattr(request, field_name, None):
             setattr(request, field_name, str(uuid.uuid4()))
+
 {% endif %}
 {% endblock %}

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2

@@ -30,6 +30,7 @@ from google.api_core import exceptions as core_exceptions
 from google.api_core import extended_operation
 {% endif %}
 from google.api_core import gapic_v1
+from {{package_path}}._compat import determine_domain, get_api_endpoint, get_default_mtls_endpoint
 {% if has_auto_populated_fields %}
 from {{package_path}}._compat import setup_request_id
 {% endif %}
@@ -146,50 +147,15 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
     """{{ service.meta.doc|rst(width=72, indent=4) }}{% if service.version|length %}
     This class implements API version {{ service.version }}.{% endif %}"""
 
-    @staticmethod
-    def _get_default_mtls_endpoint(api_endpoint) -> Optional[str]:
-        """Converts api endpoint to mTLS endpoint.
-
-        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-        Args:
-            api_endpoint (Optional[str]): the api endpoint to convert.
-        Returns:
-            Optional[str]: converted mTLS api endpoint.
-        """
-        if not api_endpoint:
-            return api_endpoint
-
-        mtls_endpoint_re = re.compile(
-            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
-        )
-
-        m = mtls_endpoint_re.match(api_endpoint)
-        if m is None:
-            # Could not parse api_endpoint; return as-is.
-            return api_endpoint
-
-        name, mtls, sandbox, googledomain = m.groups()
-        if mtls or not googledomain:
-            return api_endpoint
-
-        if sandbox:
-            return api_endpoint.replace(
-                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
-            )
-
-        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
-
     # Note: DEFAULT_ENDPOINT is deprecated. Use _DEFAULT_ENDPOINT_TEMPLATE instead.
     DEFAULT_ENDPOINT = {% if service.host %}"{{ service.host }}"{% else %}None{% endif %}
 
-    DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__(  # type: ignore
+    DEFAULT_MTLS_ENDPOINT = get_default_mtls_endpoint(
         DEFAULT_ENDPOINT
     )
 
     _DEFAULT_ENDPOINT_TEMPLATE = {% if service.host %}"{{ service.host.replace("googleapis.com", "{UNIVERSE_DOMAIN}") }}"{% else %}None{% endif %}
 
-    _DEFAULT_UNIVERSE = "googleapis.com"
 
     @staticmethod
     def _use_client_cert_effective():
@@ -394,54 +360,6 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 client_cert_source = mtls.default_client_cert_source()
         return client_cert_source
 
-    @staticmethod
-    def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtls_endpoint) -> str:
-        """Return the API endpoint used by the client.
-
-        Args:
-            api_override (str): The API endpoint override. If specified, this is always
-                the return value of this function and the other arguments are not used.
-            client_cert_source (bytes): The client certificate source used by the client.
-            universe_domain (str): The universe domain used by the client.
-            use_mtls_endpoint (str): How to use the mTLS endpoint, which depends also on the other parameters.
-                Possible values are "always", "auto", or "never".
-
-        Returns:
-            str: The API endpoint to be used by the client.
-        """
-        if api_override is not None:
-            api_endpoint = api_override
-        elif use_mtls_endpoint == "always" or (use_mtls_endpoint == "auto" and client_cert_source):
-            _default_universe = {{ service.client_name }}._DEFAULT_UNIVERSE
-            if universe_domain != _default_universe:
-                raise MutualTLSChannelError(f"mTLS is not supported in any universe other than {_default_universe}.")
-            api_endpoint = {{ service.client_name }}.DEFAULT_MTLS_ENDPOINT
-        else:
-            api_endpoint = {{ service.client_name }}._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=universe_domain)
-        return api_endpoint
-
-    @staticmethod
-    def _get_universe_domain(client_universe_domain: Optional[str], universe_domain_env: Optional[str]) -> str:
-        """Return the universe domain used by the client.
-
-        Args:
-            client_universe_domain (Optional[str]): The universe domain configured via the client options.
-            universe_domain_env (Optional[str]): The universe domain configured via the "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable.
-
-        Returns:
-            str: The universe domain to be used by the client.
-
-        Raises:
-            ValueError: If the universe domain is an empty string.
-        """
-        universe_domain = {{ service.client_name }}._DEFAULT_UNIVERSE
-        if client_universe_domain is not None:
-            universe_domain = client_universe_domain
-        elif universe_domain_env is not None:
-            universe_domain = universe_domain_env
-        if len(universe_domain.strip()) == 0:
-            raise ValueError("Universe Domain cannot be an empty string.") 
-        return universe_domain
 
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.
@@ -570,7 +488,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = {{ service.client_name }}._read_environment_variables()
         self._client_cert_source = {{ service.client_name }}._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = {{ service.client_name }}._get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = determine_domain(universe_domain_opt, self._universe_domain_env)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
@@ -601,8 +519,8 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             self._transport = cast({{ service.name }}Transport, transport)
             self._api_endpoint = self._transport.host
 
-        self._api_endpoint = (self._api_endpoint or
-            {{ service.client_name }}._get_api_endpoint(
+        self._api_endpoint = (self._api_endpoint or 
+            get_api_endpoint(
                 self._client_options.api_endpoint,
                 self._client_cert_source,
                 self._universe_domain,