Skip to content

chore(generator): use routing helper from api_core in templates #17746

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
hebaalazzeh wants to merge 3 commits into
base: feat/gapic-centralization-api-core-routing
Choose a base branch
from
Draft

chore(generator): use routing helper from api_core in templates #17746

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
05fb32e
refactor(generator): use routing helper from api_core in templates an…
hebaalazzeh Jul 16, 2026
9f58958
Merge branch 'feat/gapic-centralization-api-core-routing' into feat/g…
hebaalazzeh Jul 16, 2026
331a643
refactor: address review comments on centralized routing helpers
hebaalazzeh Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 24 additions & 56 deletions ...c-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 google.api_core.gapic_v1 import routing
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
Expand Down Expand Up @@ -143,44 +144,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_MTLS_ENDPOINT = routing.get_default_mtls_endpoint(
DEFAULT_ENDPOINT
)

Expand Down Expand Up @@ -392,30 +359,34 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
return client_cert_source

@staticmethod
def _get_api_endpoint(api_override, client_cert_source, universe_domain, use_mtls_endpoint) -> str:
def _get_api_endpoint(
api_override: Optional[str],
client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
universe_domain: str,
use_mtls_endpoint: str,
) -> str:
"""Return the API endpoint used by the client.

Args:
api_override (str): The API endpoint override. If specified, this is always
api_override (Optional[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.
client_cert_source (Union[Callable[[], Tuple[bytes, bytes]], None]): 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
return routing.get_api_endpoint(
api_override,
client_cert_source,
universe_domain,
use_mtls_endpoint,
{{ service.client_name }}._DEFAULT_UNIVERSE,
{{ service.client_name }}.DEFAULT_MTLS_ENDPOINT,
{{ service.client_name }}._DEFAULT_ENDPOINT_TEMPLATE,
)

@staticmethod
def _get_universe_domain(client_universe_domain: Optional[str], universe_domain_env: Optional[str]) -> str:
Expand All @@ -431,14 +402,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 routing.get_universe_domain(
client_universe_domain,
universe_domain_env,
{{ service.client_name }}._DEFAULT_UNIVERSE,
)

def _validate_universe_domain(self):
"""Validates client's and credentials' universe domains are consistent.
Expand Down
48 changes: 0 additions & 48 deletions .../gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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 %}
Expand Down Expand Up @@ -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),
Expand Down
102 changes: 26 additions & 76 deletions ...or/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.api_core.gapic_v1 import routing
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
Expand Down Expand Up @@ -101,38 +102,9 @@ class AssetServiceClient(metaclass=AssetServiceClientMeta):
"""Asset service definition."""

@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")
def _get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
"""Converts api endpoint to mTLS endpoint."""
return routing.get_default_mtls_endpoint(api_endpoint)

# Note: DEFAULT_ENDPOINT is deprecated. Use _DEFAULT_ENDPOINT_TEMPLATE instead.
DEFAULT_ENDPOINT = "cloudasset.googleapis.com"
Expand Down Expand Up @@ -450,53 +422,31 @@ def _get_client_cert_source(provided_cert_source, use_cert_flag):
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 = AssetServiceClient._DEFAULT_UNIVERSE
if universe_domain != _default_universe:
raise MutualTLSChannelError(f"mTLS is not supported in any universe other than {_default_universe}.")
api_endpoint = AssetServiceClient.DEFAULT_MTLS_ENDPOINT
else:
api_endpoint = AssetServiceClient._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=universe_domain)
return api_endpoint
def _get_api_endpoint(
api_override: Optional[str],
client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
universe_domain: str,
use_mtls_endpoint: str,
) -> str:
"""Return the API endpoint used by the client."""
return routing.get_api_endpoint(
api_override,
client_cert_source,
universe_domain,
use_mtls_endpoint,
AssetServiceClient._DEFAULT_UNIVERSE,
AssetServiceClient.DEFAULT_MTLS_ENDPOINT,
AssetServiceClient._DEFAULT_ENDPOINT_TEMPLATE,
)

@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 routing.get_universe_domain(
client_universe_domain,
universe_domain_env,
AssetServiceClient._DEFAULT_UNIVERSE,
)

def _validate_universe_domain(self):
"""Validates client's and credentials' universe domains are consistent.
Expand Down
35 changes: 0 additions & 35 deletions ...generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,41 +278,6 @@ def test__get_client_cert_source():
assert AssetServiceClient._get_client_cert_source(None, True) is mock_default_cert_source
assert AssetServiceClient._get_client_cert_source(mock_provided_cert_source, "true") is mock_provided_cert_source

@mock.patch.object(AssetServiceClient, "_DEFAULT_ENDPOINT_TEMPLATE", modify_default_endpoint_template(AssetServiceClient))
@mock.patch.object(AssetServiceAsyncClient, "_DEFAULT_ENDPOINT_TEMPLATE", modify_default_endpoint_template(AssetServiceAsyncClient))
def test__get_api_endpoint():
api_override = "foo.com"
mock_client_cert_source = mock.Mock()
default_universe = AssetServiceClient._DEFAULT_UNIVERSE
default_endpoint = AssetServiceClient._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=default_universe)
mock_universe = "bar.com"
mock_endpoint = AssetServiceClient._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=mock_universe)

assert AssetServiceClient._get_api_endpoint(api_override, mock_client_cert_source, default_universe, "always") == api_override
assert AssetServiceClient._get_api_endpoint(None, mock_client_cert_source, default_universe, "auto") == AssetServiceClient.DEFAULT_MTLS_ENDPOINT
assert AssetServiceClient._get_api_endpoint(None, None, default_universe, "auto") == default_endpoint
assert AssetServiceClient._get_api_endpoint(None, None, default_universe, "always") == AssetServiceClient.DEFAULT_MTLS_ENDPOINT
assert AssetServiceClient._get_api_endpoint(None, mock_client_cert_source, default_universe, "always") == AssetServiceClient.DEFAULT_MTLS_ENDPOINT
assert AssetServiceClient._get_api_endpoint(None, None, mock_universe, "never") == mock_endpoint
assert AssetServiceClient._get_api_endpoint(None, None, default_universe, "never") == default_endpoint

with pytest.raises(MutualTLSChannelError) as excinfo:
AssetServiceClient._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."


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),
Expand Down
Loading
Loading