feat(generator): gapic generator centralization routing

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

@@ -13,12 +13,158 @@ 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. #}
 
+from typing import Optional
+from urllib.parse import urlparse, urlunparse
+
+from google.auth.exceptions import MutualTLSChannelError
+from google.api_core.universe import EmptyUniverseError
+
 {% if has_auto_populated_fields %}
-from typing import Union
 import uuid
-
+from typing import Union
 import google.protobuf.message
+{% endif %}
+
+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 +209,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,46 +147,10 @@ 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_ENDPOINT
-    )
+    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 %}
 
@@ -348,7 +313,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         if client_options.api_endpoint is not None:
             api_endpoint = client_options.api_endpoint
         elif use_mtls_endpoint == "always" or (use_mtls_endpoint == "auto" and client_cert_source):
-            api_endpoint = cls.DEFAULT_MTLS_ENDPOINT
+            api_endpoint = cls.DEFAULT_MTLS_ENDPOINT or ""
         else:
             api_endpoint = cls.DEFAULT_ENDPOINT
 
@@ -394,54 +359,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 +487,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,12 +518,17 @@ 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._client_options.api_endpoint,
-                self._client_cert_source,
-                self._universe_domain,
-                self._use_mtls_endpoint))
+        self._api_endpoint = (self._api_endpoint or 
+            get_api_endpoint(
+                api_override=self._client_options.api_endpoint,
+                universe_domain=self._universe_domain,
+                default_universe={{ service.client_name }}._DEFAULT_UNIVERSE,
+                default_mtls_endpoint={{ service.client_name }}.DEFAULT_MTLS_ENDPOINT,
+                default_endpoint_template={{ service.client_name }}._DEFAULT_ENDPOINT_TEMPLATE,
+                use_mtls=self._use_mtls_endpoint == "always" or (
+                    self._use_mtls_endpoint == "auto" and self._client_cert_source
+                ),
+            ))
 
         if not transport_provided:
             {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2121): Remove this condition when async rest is GA. #}

packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2

@@ -156,22 +156,6 @@ def set_event_loop():
             asyncio.set_event_loop(None)
 
 
-def test__get_default_mtls_endpoint():
-    api_endpoint = "example.googleapis.com"
-    api_mtls_endpoint = "example.mtls.googleapis.com"
-    sandbox_endpoint = "example.sandbox.googleapis.com"
-    sandbox_mtls_endpoint = "example.mtls.sandbox.googleapis.com"
-    non_googleapi = "api.example.com"
-    custom_endpoint = ".custom"
-
-    assert {{ service.client_name }}._get_default_mtls_endpoint(None) is None
-    assert {{ service.client_name }}._get_default_mtls_endpoint(api_endpoint) == api_mtls_endpoint
-    assert {{ service.client_name }}._get_default_mtls_endpoint(api_mtls_endpoint) == api_mtls_endpoint
-    assert {{ service.client_name }}._get_default_mtls_endpoint(sandbox_endpoint) == sandbox_mtls_endpoint
-    assert {{ service.client_name }}._get_default_mtls_endpoint(sandbox_mtls_endpoint) == sandbox_mtls_endpoint
-    assert {{ service.client_name }}._get_default_mtls_endpoint(non_googleapi) == non_googleapi
-    assert {{ service.client_name }}._get_default_mtls_endpoint(custom_endpoint) == custom_endpoint
-
 def test__read_environment_variables():
     assert {{ service.client_name }}._read_environment_variables() == (False, "auto", None)
 
@@ -313,29 +297,7 @@ def test__get_client_cert_source():
             assert {{ service.client_name }}._get_client_cert_source(None, True) is mock_default_cert_source
             assert {{ service.client_name }}._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source
 
-@mock.patch.object({{ service.client_name }}, "_DEFAULT_ENDPOINT_TEMPLATE", modify_default_endpoint_template({{ service.client_name }}))
-{% if 'grpc' in opts.transport %}
-@mock.patch.object({{ service.async_client_name }}, "_DEFAULT_ENDPOINT_TEMPLATE", modify_default_endpoint_template({{ service.async_client_name }}))
-{% endif %}
-def test__get_api_endpoint():
-    api_override = "foo.com"
-    mock_client_cert_source = mock.Mock()
-    default_universe = {{ service.client_name }}._DEFAULT_UNIVERSE
-    default_endpoint = {{ service.client_name }}._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=default_universe)
-    mock_universe = "bar.com"
-    mock_endpoint = {{ service.client_name }}._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=mock_universe)
-
-    assert {{ service.client_name }}._get_api_endpoint(api_override, mock_client_cert_source, default_universe, "always") == api_override
-    assert {{ service.client_name }}._get_api_endpoint(None, mock_client_cert_source, default_universe, "auto") == {{ service.client_name }}.DEFAULT_MTLS_ENDPOINT
-    assert {{ service.client_name }}._get_api_endpoint(None, None, default_universe, "auto") == default_endpoint
-    assert {{ service.client_name }}._get_api_endpoint(None, None, default_universe, "always") == {{ service.client_name }}.DEFAULT_MTLS_ENDPOINT
-    assert {{ service.client_name }}._get_api_endpoint(None, mock_client_cert_source, default_universe, "always") == {{ service.client_name }}.DEFAULT_MTLS_ENDPOINT
-    assert {{ service.client_name }}._get_api_endpoint(None, None, mock_universe, "never") == mock_endpoint
-    assert {{ service.client_name }}._get_api_endpoint(None, None, default_universe, "never") == default_endpoint
 
-    with pytest.raises(MutualTLSChannelError) as excinfo:
-        {{ service.client_name }}._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
-    assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
 {% if service.version %}
 {% for method in service.methods.values() %}{% with method_name = method.name|snake_case %}
@@ -384,17 +346,7 @@ def test_{{ method_name }}_api_version_header(transport_name):
 {% endfor %}
 {% endif %}{# service.version #}
 
-def test__get_universe_domain():
-    client_universe_domain = "foo.com"
-    universe_domain_env = "bar.com"
-
-    assert {{ service.client_name }}._get_universe_domain(client_universe_domain, universe_domain_env) == client_universe_domain
-    assert {{ service.client_name }}._get_universe_domain(None, universe_domain_env) == universe_domain_env
-    assert {{ service.client_name }}._get_universe_domain(None, None) == {{ service.client_name }}._DEFAULT_UNIVERSE
 
-    with pytest.raises(ValueError) as excinfo:
-        {{ service.client_name }}._get_universe_domain("", None)
-    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
 
 @pytest.mark.parametrize("error_code,cred_info_json,show_cred_info", [
     (401, CRED_INFO_JSON, True),