chore(api_core): move universe and endpoint routing logic to gapic_v1 public helpers

packages/google-api-core/google/api_core/gapic_v1/__init__.py

@@ -25,9 +25,10 @@
 # Older Python versions safely ignore this variable.
 __lazy_modules__: Set[str] = {
     "google.api_core.gapic_v1.client_info",
+    "google.api_core.gapic_v1.routing",
     "google.api_core.gapic_v1.routing_header",
 }
-__all__ = ["client_info", "routing_header"]
+__all__ = ["client_info", "routing", "routing_header"]
 
 if _has_grpc:
     __lazy_modules__.update(
@@ -41,6 +42,7 @@
 
 from google.api_core.gapic_v1 import (  # noqa: E402
     client_info,
+    routing,
     routing_header,
 )
 

packages/google-api-core/google/api_core/gapic_v1/routing.py

@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Helpers for routing and endpoint resolution."""
+
+from typing import Any, Optional
+
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+
+
+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.
+
+    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:
+        return api_endpoint
+
+    if api_endpoint.endswith(".sandbox.googleapis.com"):
+        # len(".sandbox.googleapis.com") == 23
+        return api_endpoint[:-23] + ".mtls.sandbox.googleapis.com"
+
+    if api_endpoint.endswith(".googleapis.com"):
+        # len(".googleapis.com") == 15
+        return api_endpoint[:-15] + ".mtls.googleapis.com"
+
+    return api_endpoint
+
+
+def get_api_endpoint(
+    api_override: Optional[str],
+    client_cert_source: Optional[Any],
+    universe_domain: str,
+    use_mtls_endpoint: str,
+    default_universe: str,
+    default_mtls_endpoint: Optional[str],
+    default_endpoint_template: str,
+) -> str:
+    """Return the API endpoint used by the client.
+
+    Args:
+        api_override (Optional[str]): The API endpoint override. If specified,
+            this is always returned.
+        client_cert_source (Optional[Any]): 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. Possible values
+            are "always", "auto", or "never".
+        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}`.
+
+    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
+    elif use_mtls_endpoint == "always" or (
+        use_mtls_endpoint == "auto" and client_cert_source
+    ):
+        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(
+    client_universe_domain: Optional[str],
+    universe_domain_env: Optional[str],
+    default_universe: str,
+) -> str:
+    """Return the universe domain used by the client.
+
+    Args:
+        client_universe_domain (Optional[str]): The universe domain configured
+            via client options.
+        universe_domain_env (Optional[str]): The universe domain configured
+            via environment variable.
+        default_universe (str): The default universe domain.
+
+    Returns:
+        str: The universe domain to be used by the client.
+
+    Raises:
+        ValueError: If the resolved universe domain is an empty string.
+    """
+    if client_universe_domain is not None:
+        universe_domain = client_universe_domain.strip()
+    elif universe_domain_env is not None:
+        universe_domain = universe_domain_env.strip()
+    else:
+        universe_domain = default_universe
+
+    if not universe_domain:
+        raise ValueError("Universe Domain cannot be an empty string.")
+    return universe_domain

packages/google-api-core/tests/conftest.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from unittest import mock
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def mock_mtls_env():
+    """Autouse session-scoped fixture to isolate unit tests from workstation mTLS environments."""
+    with mock.patch.dict(
+        os.environ,
+        {
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false",
+            "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false",
+        },
+    ):
+        yield

packages/google-api-core/tests/unit/gapic/test_routing.py

@@ -0,0 +1,201 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+
+import pytest
+
+from google.auth.exceptions import MutualTLSChannelError
+
+from google.api_core.gapic_v1.routing import (
+    get_api_endpoint,
+    get_default_mtls_endpoint,
+    get_universe_domain,
+)
+
+
+class MockClient:
+    _DEFAULT_UNIVERSE = "googleapis.com"
+    DEFAULT_MTLS_ENDPOINT = "foo.mtls.googleapis.com"
+    _DEFAULT_ENDPOINT_TEMPLATE = "foo.{UNIVERSE_DOMAIN}"
+
+
+def test_get_default_mtls_endpoint():
+    # Test valid API endpoints
+    assert get_default_mtls_endpoint("foo.googleapis.com") == "foo.mtls.googleapis.com"
+    assert (
+        get_default_mtls_endpoint("foo.sandbox.googleapis.com")
+        == "foo.mtls.sandbox.googleapis.com"
+    )
+
+    # Test endpoints that shouldn't be converted
+    assert (
+        get_default_mtls_endpoint("foo.mtls.googleapis.com")
+        == "foo.mtls.googleapis.com"
+    )
+    assert get_default_mtls_endpoint("foo.com") == "foo.com"
+
+    # Test empty/None endpoints
+    assert get_default_mtls_endpoint("") == ""
+    assert get_default_mtls_endpoint(None) is None
+
+
+def test__get_api_endpoint():
+    api_override = "foo.com"
+    mock_client_cert_source = mock.Mock()
+    default_universe = MockClient._DEFAULT_UNIVERSE
+    default_endpoint = MockClient._DEFAULT_ENDPOINT_TEMPLATE.format(
+        UNIVERSE_DOMAIN=default_universe
+    )
+    mock_universe = "bar.com"
+    mock_endpoint = MockClient._DEFAULT_ENDPOINT_TEMPLATE.format(
+        UNIVERSE_DOMAIN=mock_universe
+    )
+
+    assert (
+        get_api_endpoint(
+            api_override,
+            mock_client_cert_source,
+            default_universe,
+            "always",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == api_override
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            mock_client_cert_source,
+            default_universe,
+            "auto",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == MockClient.DEFAULT_MTLS_ENDPOINT
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            None,
+            default_universe,
+            "auto",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == default_endpoint
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            None,
+            default_universe,
+            "always",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == MockClient.DEFAULT_MTLS_ENDPOINT
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            mock_client_cert_source,
+            default_universe,
+            "always",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == MockClient.DEFAULT_MTLS_ENDPOINT
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            None,
+            mock_universe,
+            "never",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == mock_endpoint
+    )
+    assert (
+        get_api_endpoint(
+            None,
+            None,
+            default_universe,
+            "never",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+        == default_endpoint
+    )
+
+    with pytest.raises(MutualTLSChannelError) as excinfo:
+        get_api_endpoint(
+            None,
+            mock_client_cert_source,
+            mock_universe,
+            "auto",
+            MockClient._DEFAULT_UNIVERSE,
+            MockClient.DEFAULT_MTLS_ENDPOINT,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+    assert (
+        str(excinfo.value)
+        == "mTLS is not supported in any universe other than googleapis.com."
+    )
+
+    with pytest.raises(ValueError) as excinfo:
+        get_api_endpoint(
+            None,
+            mock_client_cert_source,
+            default_universe,
+            "always",
+            MockClient._DEFAULT_UNIVERSE,
+            None,
+            MockClient._DEFAULT_ENDPOINT_TEMPLATE,
+        )
+    assert str(excinfo.value) == "mTLS endpoint is not available."
+
+
+def test__get_universe_domain():
+    client_universe_domain = "foo.com"
+    universe_domain_env = "bar.com"
+
+    assert (
+        get_universe_domain(
+            client_universe_domain, universe_domain_env, MockClient._DEFAULT_UNIVERSE
+        )
+        == client_universe_domain
+    )
+    assert (
+        get_universe_domain(None, universe_domain_env, MockClient._DEFAULT_UNIVERSE)
+        == universe_domain_env
+    )
+    assert (
+        get_universe_domain(None, None, MockClient._DEFAULT_UNIVERSE)
+        == MockClient._DEFAULT_UNIVERSE
+    )
+
+    with pytest.raises(ValueError) as excinfo:
+        get_universe_domain("", None, MockClient._DEFAULT_UNIVERSE)
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."