feat(api_core): add universe and endpoint routing logic to gapic_v1 public helpers

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

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

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

@@ -0,0 +1,133 @@
+# 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 typing import Optional
+from urllib.parse import urlparse, urlunparse
+
+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.
+    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()
+    if lowered_host.endswith(".sandbox.googleapis.com"):
+        new_host = host[:-23] + ".mtls.sandbox.googleapis.com"
+    elif lowered_host.endswith(".googleapis.com"):
+        new_host = host[:-15] + ".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(
+    universe_domain: Optional[str],
+    default_universe: str = "googleapis.com",
+) -> str:
+    """Return the universe domain used by the client.
+
+    Args:
+        universe_domain (Optional[str]): The configured universe domain.
+        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.
+    """
+    resolved = (
+        universe_domain.strip() if universe_domain is not None else default_universe
+    )
+
+    if not resolved:
+        raise ValueError("Universe Domain cannot be an empty string.")
+    return resolved

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

@@ -21,8 +21,8 @@
 if they are not already set.
 """
 
-from typing import Union
 import uuid
+from typing import Union
 
 import google.protobuf.message
 
@@ -65,7 +65,7 @@ def setup_request_id(
                 setattr(request, field_name, str(uuid.uuid4()))
         except (AttributeError, ValueError):
             # Proto-plus messages or other objects
-            if not getattr(request, field_name, None):
+            if getattr(request, field_name, None) is None:
                 setattr(request, field_name, str(uuid.uuid4()))
     else:
         if not getattr(request, field_name, None):

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

@@ -0,0 +1,31 @@
+# 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_client_utils.py

@@ -0,0 +1,179 @@
+# 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 pytest
+from google.auth.exceptions import MutualTLSChannelError
+
+from google.api_core.gapic_v1.client_utils import (
+    get_api_endpoint,
+    get_default_mtls_endpoint,
+    get_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 case-insensitivity
+    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 valid API endpoints with schemes
+    assert (
+        get_default_mtls_endpoint("https://foo.googleapis.com")
+        == "https://foo.mtls.googleapis.com"
+    )
+    assert (
+        get_default_mtls_endpoint("http://foo.googleapis.com:8080/v1")
+        == "http://foo.mtls.googleapis.com:8080/v1"
+    )
+
+    # Test valid API endpoints with ports
+    assert (
+        get_default_mtls_endpoint("foo.googleapis.com:443")
+        == "foo.mtls.googleapis.com:443"
+    )
+    assert (
+        get_default_mtls_endpoint("foo.sandbox.googleapis.com:443")
+        == "foo.mtls.sandbox.googleapis.com:443"
+    )
+    # Test case-insensitivity with ports
+    assert (
+        get_default_mtls_endpoint("foo.GoogleAPIs.com:443")
+        == "foo.mtls.googleapis.com:443"
+    )
+    assert (
+        get_default_mtls_endpoint("foo.Sandbox.GoogleAPIs.com:443")
+        == "foo.mtls.sandbox.googleapis.com:443"
+    )
+
+    # 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"
+    assert get_default_mtls_endpoint("foo.com:8080") == "foo.com:8080"
+
+    # Test empty/None endpoints
+    assert get_default_mtls_endpoint("") == ""
+    assert get_default_mtls_endpoint(None) is None
+
+
+@pytest.mark.parametrize(
+    "api_override,universe_domain,default_universe,default_mtls_endpoint,default_endpoint_template,use_mtls,expected",
+    [
+        (
+            "foo.com",
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            "foo.com",
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            "foo.mtls.googleapis.com",
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            False,
+            "foo.googleapis.com",
+        ),
+        (
+            None,
+            "bar.com",
+            "googleapis.com",
+            "foo.mtls.googleapis.com",
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            MutualTLSChannelError,
+        ),
+        (
+            None,
+            "googleapis.com",
+            "googleapis.com",
+            None,
+            "foo.{UNIVERSE_DOMAIN}",
+            True,
+            ValueError,
+        ),
+    ],
+)
+def test_get_api_endpoint(
+    api_override,
+    universe_domain,
+    default_universe,
+    default_mtls_endpoint,
+    default_endpoint_template,
+    use_mtls,
+    expected,
+):
+    if isinstance(expected, type) and issubclass(expected, Exception):
+        with pytest.raises(expected):
+            get_api_endpoint(
+                api_override,
+                universe_domain,
+                default_universe,
+                default_mtls_endpoint,
+                default_endpoint_template,
+                use_mtls,
+            )
+    else:
+        assert (
+            get_api_endpoint(
+                api_override,
+                universe_domain,
+                default_universe,
+                default_mtls_endpoint,
+                default_endpoint_template,
+                use_mtls,
+            )
+            == expected
+        )
+
+
+def test_get_universe_domain():
+    # When universe_domain is provided
+    assert get_universe_domain("foo.com", "default.com") == "foo.com"
+    assert get_universe_domain("  foo.com  ", "default.com") == "foo.com"
+
+    # When universe_domain is None, falls back to default_universe
+    assert get_universe_domain(None, "default.com") == "default.com"
+
+    # ValueError raised when resolved value is empty string
+    with pytest.raises(ValueError) as excinfo:
+        get_universe_domain("", "default.com")
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."
+
+    with pytest.raises(ValueError) as excinfo:
+        get_universe_domain("   ", "default.com")
+    assert str(excinfo.value) == "Universe Domain cannot be an empty string."