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

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,76 @@
+# {% include '_license.j2' %}
+
+import re
+from typing import Optional, Callable, Tuple, Union
+from google.auth.exceptions import MutualTLSChannelError
+
+try:
+ from google.api_core.gapic_v1.client_utils import (
+ get_default_mtls_endpoint,
+ get_api_endpoint,
+ get_universe_domain,
+ )
+except ImportError:
+ # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+ def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+ """Converts api endpoint to mTLS 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_api_endpoint(
+ api_override: Optional[str],
+ client_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+ universe_domain: str,
+ use_mtls_endpoint: str,
+ default_universe: str,
+ default_mtls_endpoint: str,
+ default_endpoint_template: str,
+ ) -> str:
+ """Return the API endpoint 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):
+ if universe_domain != default_universe:
+ raise MutualTLSChannelError(
+ f"mTLS is not supported in any universe other than {default_universe}."
+ )
+ api_endpoint = default_mtls_endpoint
+ else:
+ api_endpoint = default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+ return api_endpoint
+
+ 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."""
+ universe_domain = 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

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
@@ -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 = client_utils.get_default_mtls_endpoint(
  DEFAULT_ENDPOINT
  )
 
@@ -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 client_utils.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:
@@ -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 client_utils.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.

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),