feat(generator): add fallback compatibility imports for google-api-core helpers

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2

@@ -442,7 +442,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables()
         self._client_cert_source = {{ service.client_name }}._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe={{ service.client_name }}._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.

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

@@ -69,175 +69,186 @@ def read_environment_variables():
 
 DEFAULT_UNIVERSE = "googleapis.com"
 
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    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()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".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_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.
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    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)
 
-    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()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".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.
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
+        )
 
-    Returns:
-        str: The API endpoint to be used by the client.
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
 
-    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}."
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
+                use_integers_for_enums=rest_numeric_enums,
             )
-        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(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
 
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
-        )
-
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
-                use_integers_for_enums=rest_numeric_enums,
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json

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

@@ -491,7 +491,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = read_environment_variables()
         self._client_cert_source = AssetServiceClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=AssetServiceClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.