googleapis · hebaalazzeh · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026
@@ -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.

@@ -0,0 +1,270 @@
+# {% include '_license.j2' %}
+
+"""A compatibility module for older versions of google-api-core."""
+
+import functools
+import json
+import operator
+import os
+import re
+import uuid
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from google.auth.exceptions import MutualTLSChannelError
+import google.protobuf.message
+
+
+try:
+    from google.api_core.universe import (
+        get_default_mtls_endpoint,
+        get_api_endpoint,
+        get_universe_domain,
+    )
+except ImportError:
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): 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: Optional[str],
+        default_endpoint_template: str,
+    ) -> Optional[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
+
+
+try:
+    from google.api_core.gapic_v1.config import (
+        use_client_cert_effective,
+        get_client_cert_source,
+        read_environment_variables,
+    )
+except ImportError:
+    from google.auth.transport import mtls  # type: ignore
+
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+
+    def use_client_cert_effective() -> bool:
+        """Returns whether client certificate should be used for mTLS."""
+        if hasattr(mtls, "should_use_client_cert"):
+            return mtls.should_use_client_cert()
+        else:
+            use_client_cert_str = os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
+            if use_client_cert_str not in ("true", "false"):
+                raise ValueError(
+                    "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be"
+                    " either `true` or `false`"
+                )
+            return use_client_cert_str == "true"
+
+    def get_client_cert_source(
+        provided_cert_source: Optional[Callable[[], Tuple[bytes, bytes]]],
+        use_cert_flag: bool,
+    ) -> Optional[Callable[[], Tuple[bytes, bytes]]]:
+        """Return the client cert source to be used by the client."""
+        client_cert_source = None
+        if use_cert_flag:
+            if provided_cert_source:
+                client_cert_source = provided_cert_source
+            elif (
+                hasattr(mtls, "has_default_client_cert_source")
+                and mtls.has_default_client_cert_source()
+            ):
+                client_cert_source = mtls.default_client_cert_source()
+            else:
+                raise ValueError(
+                    "Client certificate is required for mTLS, but no client certificate source was provided or found."
+                )
+        return client_cert_source
+
+    def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
+        """Returns the environment variables used by the client."""
+        use_client_cert = use_client_cert_effective()
+        use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower()
+        universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN")
+        if use_mtls_endpoint not in ("auto", "never", "always"):
+            raise MutualTLSChannelError(
+                "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` "
+                "must be `never`, `auto` or `always`"
+            )
+        return use_client_cert, use_mtls_endpoint, universe_domain_env
+
+
+try:
+    from google.api_core.gapic_v1.request import setup_request_id  # type: ignore
+except ImportError:
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove this fallback when google-api-core >= 2.26.0 is the minimum required version.
+    def setup_request_id(request, field_name: str, is_proto3_optional: bool):
+        """Populate a UUID4 field in the request if it is not already set.
+
+        Args:
+            request (Union[google.protobuf.message.Message, dict]): The request object.
+            field_name (str): The name of the field to populate.
+            is_proto3_optional (bool): Whether the field is proto3 optional.
+        """
+        request_id_val = str(uuid.uuid4())
+        if request is None:
+            return
+
+        if isinstance(request, dict):
+            if is_proto3_optional:
+                if field_name not in request or request[field_name] is None:
+                    request[field_name] = request_id_val
+            elif not request.get(field_name):
+                request[field_name] = request_id_val
+            return
+
+        if is_proto3_optional:
+            try:
+                # Pure protobuf messages
+                if not request.HasField(field_name):
+                    setattr(request, field_name, request_id_val)
+            except (AttributeError, ValueError):
+                # Proto-plus messages or other objects
+                if getattr(request, field_name, None) is None:
+                    setattr(request, field_name, request_id_val)
+        else:
+            if not getattr(request, field_name, None):
+                setattr(request, field_name, request_id_val)
+
+
+try:
+    from google.api_core.rest_helpers import (
+        flatten_query_params,
+        transcode_request,
+    )
+except ImportError:  # pragma: NO COVER
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    from google.protobuf import json_format  # type: ignore
+    from google.api_core import path_template  # type: ignore
+
+    def flatten_query_params(obj, strict=False):  # pragma: NO COVER
+        if obj is not None and not isinstance(obj, dict):
+            raise TypeError("flatten_query_params must be called with dict object")
+        return _flatten(obj, key_path=[], strict=strict)
+
+    def _flatten(obj, key_path, strict=False):  # pragma: NO COVER
+        if obj is None:
+            return []
+        if isinstance(obj, dict):
+            return _flatten_dict(obj, key_path=key_path, strict=strict)
+        if isinstance(obj, list):
+            return _flatten_list(obj, key_path=key_path, strict=strict)
+        return _flatten_value(obj, key_path=key_path, strict=strict)
+
+    def _is_primitive_value(obj):  # pragma: NO COVER
+        if obj is None:
+            return False
+        if isinstance(obj, (list, dict)):
+            raise ValueError("query params may not contain repeated dicts or lists")
+        return True
+
+    def _flatten_value(obj, key_path, strict=False):  # pragma: NO COVER
+        return [(".".join(key_path), _canonicalize(obj, strict=strict))]
+
+    def _flatten_dict(obj, key_path, strict=False):  # pragma: NO COVER
+        items = (
+            _flatten(value, key_path=key_path + [key], strict=strict)
+            for key, value in obj.items()
+        )
+        return functools.reduce(operator.concat, items, [])
+
+    def _flatten_list(elems, key_path, strict=False):  # pragma: NO COVER
+        items = (
+            _flatten_value(elem, key_path=key_path, strict=strict)
+            for elem in elems
+            if _is_primitive_value(elem)
+        )
+        return functools.reduce(operator.concat, items, [])
+
+    def _canonicalize(obj, strict=False):  # pragma: NO COVER
+        if strict:
+            value = str(obj)
+            if isinstance(obj, bool):
+                value = value.lower()
+            return value
+        return obj
+
+    def transcode_request(  # pragma: NO COVER
+        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]]:
+        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,
+            ))
+
+        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"
+
+        return transcoded_request, body_json, query_params_json
@@ -198,22 +198,23 @@ def _get_http_options():
         service: The service.
         is_async (bool): Used to determine the code path i.e. whether for sync or async call.
         is_request_message_proto_plus_type (bool): Used to determine whether the request message is a proto-plus type. #}
-{% macro rest_call_method_common(body_spec, method_name, service, is_async=False, is_request_message_proto_plus_type=False) %}
+{% macro rest_call_method_common(body_spec, method_name, service, is_async=False, is_request_message_proto_plus_type=False, rest_numeric_enums=False) %}
 {% set service_name = service.name %}
 {% set await_prefix = "await " if is_async else "" %}
 {% set async_class_prefix = "Async" if is_async else "" %}
 
     http_options = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_http_options()
-    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2274): Add debug log before intercepting a request #} 
     request, metadata = {{ await_prefix }}self._interceptor.pre_{{ method_name|snake_case }}(request, metadata)
-    transcoded_request = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_transcoded_request(http_options, request)
-
-    {% if body_spec %}
-    body = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_request_body_json(transcoded_request)
-    {% endif %}{# body_spec #}
-
-    # Jsonify the query params
-    query_params = _Base{{ service_name }}RestTransport._Base{{method_name}}._get_query_params_json(transcoded_request)
+    transcoded_request, body, query_params = rest_helpers.transcode_request(
+        http_options,
+        request,
+        required_fields_default_values=getattr(
+            _Base{{ service_name }}RestTransport._Base{{method_name}},
+            "__REQUIRED_FIELDS_DEFAULT_VALUES",
+            None,
+        ),
+        rest_numeric_enums={{ rest_numeric_enums }},
+    )
 
     if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(logging.DEBUG):  # pragma: NO COVER
         request_url = "{host}{uri}".format(host=self._host, uri=transcoded_request['uri'])

@@ -25,27 +25,7 @@
 
         {{ shared_macros.http_options_method(api.mixin_http_options["{}".format(name)])|indent(8)}}
 
-        @staticmethod
-        def _get_transcoded_request(http_options, request):
-            request_kwargs = json_format.MessageToDict(request)
-            transcoded_request = path_template.transcode(
-                http_options, **request_kwargs)
-            return transcoded_request
-
-        {% set body_spec = api.mixin_http_options["{}".format(name)][0].body %}
-        {%- if body_spec %}
-
-        @staticmethod
-        def _get_request_body_json(transcoded_request):
-            body = json.dumps(transcoded_request['body'])
-            return body
-
-        {%- endif %} {# body_spec #}
-
-        @staticmethod
-        def _get_query_params_json(transcoded_request):
-            query_params = json.loads(json.dumps(transcoded_request['query_params']))
-            return query_params
+        pass
 
     {% endfor %}
     {% endif %} {# rest in opts.transport #}
@@ -10,7 +10,8 @@ from google.auth.transport.requests import AuthorizedSession  # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
-from google.api_core import rest_helpers
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}} import _compat as rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
 import google.protobuf
@@ -245,7 +246,7 @@ class {{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             {% endif %}
             """
 
-            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, False, method.input.ident.is_proto_plus_type)|indent(8) }}
+            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, False, method.input.ident.is_proto_plus_type, opts.rest_numeric_enums)|indent(8) }}
 
             {% if not method.void %}
             # Return the response

@@ -32,7 +32,8 @@ from google.iam.v1 import policy_pb2  # type: ignore
 from google.cloud.location import locations_pb2 # type: ignore
 {% endif %}
 from google.api_core import retry_async as retries
-from google.api_core import rest_helpers
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}} import _compat as rest_helpers
 from google.api_core import rest_streaming_async  # type: ignore
 import google.protobuf
 
@@ -203,7 +204,7 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             {% endif %}
             """
 
-            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, True, method.input.ident.is_proto_plus_type)|indent(8) }}
+            {{ shared_macros.rest_call_method_common(body_spec, method.name, service, True, method.input.ident.is_proto_plus_type, opts.rest_numeric_enums)|indent(8) }}
 
             {% if not method.void %}
             # Return the response