feat(api-core,generator): centralize endpoint routing and universe domain resolution

packages/gapic-generator/.bazelignore

@@ -0,0 +1 @@
+api-common-protos

packages/gapic-generator/gapic/generator/generator.py

@@ -120,7 +120,7 @@ def get_response(self, api_schema: api.API, opts: Options) -> CodeGeneratorRespo
         for template_name in client_templates:
             # Quick check: Skip "private" templates.
             filename = template_name.split("/")[-1]
-            if filename.startswith("_") and filename != "__init__.py.j2":
+            if filename.startswith("_") and filename not in ("__init__.py.j2", "_compat.py.j2"):
                 continue
 
             # Append to the output files dictionary.

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

@@ -0,0 +1,24 @@
+{% include '_license.j2' %}
+
+from typing import Optional
+
+try:
+    from google.api_core.gapic_v1.client_utils import (  # type: ignore
+        get_universe_domain,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        for potential_universe in potential_universes:
+            if potential_universe is not None:
+                universe_domain = potential_universe
+                break
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain

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}} import _compat as client_utils
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -431,14 +432,11 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         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
+        return client_utils.get_universe_domain(
+            client_universe_domain,
+            universe_domain_env,
+            default_universe={{ service.client_name }}._DEFAULT_UNIVERSE,
+        )
 
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.

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

@@ -384,18 +384,6 @@ 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),
     (403, CRED_INFO_JSON, True),

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py

@@ -0,0 +1,37 @@
+# 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
+
+try:
+    from google.api_core.gapic_v1.client_utils import (  # type: ignore
+        get_universe_domain,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        for potential_universe in potential_universes:
+            if potential_universe is not None:
+                universe_domain = potential_universe
+                break
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py

@@ -27,6 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.cloud.asset_v1 import _compat as client_utils
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -477,26 +478,12 @@ def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtl
 
     @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 = AssetServiceClient._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
+        """Return the universe domain used by the client."""
+        return client_utils.get_universe_domain(
+            client_universe_domain,
+            universe_domain_env,
+            default_universe=AssetServiceClient._DEFAULT_UNIVERSE,
+        )
 
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.

packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py

@@ -301,18 +301,6 @@ def test__get_api_endpoint():
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
 
-def test__get_universe_domain():
-    client_universe_domain = "foo.com"
-    universe_domain_env = "bar.com"
-
-    assert AssetServiceClient._get_universe_domain(client_universe_domain, universe_domain_env) == client_universe_domain
-    assert AssetServiceClient._get_universe_domain(None, universe_domain_env) == universe_domain_env
-    assert AssetServiceClient._get_universe_domain(None, None) == AssetServiceClient._DEFAULT_UNIVERSE
-
-    with pytest.raises(ValueError) as excinfo:
-        AssetServiceClient._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),
     (403, CRED_INFO_JSON, True),

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py

@@ -0,0 +1,37 @@
+# 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
+
+try:
+    from google.api_core.gapic_v1.client_utils import (  # type: ignore
+        get_universe_domain,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        for potential_universe in potential_universes:
+            if potential_universe is not None:
+                universe_domain = potential_universe
+                break
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain

packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py

@@ -27,6 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.iam.credentials_v1 import _compat as client_utils
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -414,26 +415,12 @@ def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtl
 
     @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 = IAMCredentialsClient._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
+        """Return the universe domain used by the client."""
+        return client_utils.get_universe_domain(
+            client_universe_domain,
+            universe_domain_env,
+            default_universe=IAMCredentialsClient._DEFAULT_UNIVERSE,
+        )
 
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.

packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py

@@ -291,18 +291,6 @@ def test__get_api_endpoint():
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
 
-def test__get_universe_domain():
-    client_universe_domain = "foo.com"
-    universe_domain_env = "bar.com"
-
-    assert IAMCredentialsClient._get_universe_domain(client_universe_domain, universe_domain_env) == client_universe_domain
-    assert IAMCredentialsClient._get_universe_domain(None, universe_domain_env) == universe_domain_env
-    assert IAMCredentialsClient._get_universe_domain(None, None) == IAMCredentialsClient._DEFAULT_UNIVERSE
-
-    with pytest.raises(ValueError) as excinfo:
-        IAMCredentialsClient._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),
     (403, CRED_INFO_JSON, True),

packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py

@@ -0,0 +1,37 @@
+# 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
+
+try:
+    from google.api_core.gapic_v1.client_utils import (  # type: ignore
+        get_universe_domain,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client."""
+        universe_domain = default_universe
+        for potential_universe in potential_universes:
+            if potential_universe is not None:
+                universe_domain = potential_universe
+                break
+        if len(universe_domain.strip()) == 0:
+            raise ValueError("Universe Domain cannot be an empty string.") 
+        return universe_domain

packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py

@@ -27,6 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.cloud.eventarc_v1 import _compat as client_utils
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials             # type: ignore
 from google.auth.transport import mtls                            # type: ignore
@@ -597,26 +598,12 @@ def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtl
 
     @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 = EventarcClient._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
+        """Return the universe domain used by the client."""
+        return client_utils.get_universe_domain(
+            client_universe_domain,
+            universe_domain_env,
+            default_universe=EventarcClient._DEFAULT_UNIVERSE,
+        )
 
     def _validate_universe_domain(self):
         """Validates client's and credentials' universe domains are consistent.

packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py

@@ -322,18 +322,6 @@ def test__get_api_endpoint():
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
 
-def test__get_universe_domain():
-    client_universe_domain = "foo.com"
-    universe_domain_env = "bar.com"
-
-    assert EventarcClient._get_universe_domain(client_universe_domain, universe_domain_env) == client_universe_domain
-    assert EventarcClient._get_universe_domain(None, universe_domain_env) == universe_domain_env
-    assert EventarcClient._get_universe_domain(None, None) == EventarcClient._DEFAULT_UNIVERSE
-
-    with pytest.raises(ValueError) as excinfo:
-        EventarcClient._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),
     (403, CRED_INFO_JSON, True),