googleapis · hebaalazzeh · Jul 18, 2026 · Jul 18, 2026 · Jul 20, 2026 · Jul 20, 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,107 @@
+# {% include '_license.j2' %}
+
+from typing import Any, Dict, List, Optional, Tuple
+
+try:
+    from google.api_core import rest_helpers
+    # Trigger fallback if rest_helpers is an older version missing 'transcode'
+    if not hasattr(rest_helpers, "transcode"):
+        raise ImportError
+except ImportError:
+    # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
+    import functools
+    import json
+    import operator
+    from google.protobuf import json_format  # type: ignore
+    from google.api_core import path_template  # type: ignore
+
+    class _FallbackRestHelpers:
+        @staticmethod
+        def flatten_query_params(obj, strict=False):
+            if obj is not None and not isinstance(obj, dict):
+                raise TypeError("flatten_query_params must be called with dict object")
+            return _FallbackRestHelpers._flatten(obj, key_path=[], strict=strict)
+
+        @staticmethod
+        def _flatten(obj, key_path, strict=False):
+            if obj is None:
+                return []
+            if isinstance(obj, dict):
+                return _FallbackRestHelpers._flatten_dict(obj, key_path=key_path, strict=strict)
+            if isinstance(obj, list):
+                return _FallbackRestHelpers._flatten_list(obj, key_path=key_path, strict=strict)
+            return _FallbackRestHelpers._flatten_value(obj, key_path=key_path, strict=strict)
+
+        @staticmethod
+        def _is_primitive_value(obj):
+            if obj is None:
+                return False
+            if isinstance(obj, (list, dict)):
+                raise ValueError("query params may not contain repeated dicts or lists")
+            return True
+
+        @staticmethod
+        def _flatten_value(obj, key_path, strict=False):
+            return [(".".join(key_path), _FallbackRestHelpers._canonicalize(obj, strict=strict))]
+
+        @staticmethod
+        def _flatten_dict(obj, key_path, strict=False):
+            items = (
+                _FallbackRestHelpers._flatten(value, key_path=key_path + [key], strict=strict)
+                for key, value in obj.items()
+            )
+            return functools.reduce(operator.concat, items, [])
+
+        @staticmethod
+        def _flatten_list(elems, key_path, strict=False):
+            items = (
+                _FallbackRestHelpers._flatten_value(elem, key_path=key_path, strict=strict)
+                for elem in elems
+                if _FallbackRestHelpers._is_primitive_value(elem)
+            )
+            return functools.reduce(operator.concat, items, [])
+
+        @staticmethod
+        def _canonicalize(obj, strict=False):
+            if strict:
+                value = str(obj)
+                if isinstance(obj, bool):
+                    value = value.lower()
+                return value
+            return obj
+
+        @staticmethod
+        def transcode(
+            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
+
+    rest_helpers = _FallbackRestHelpers
@@ -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(
+        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,7 @@ 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
+from .. import _compat as rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
 import google.protobuf
@@ -245,7 +245,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,7 @@ 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
+from .. import _compat as rest_helpers
 from google.api_core import rest_streaming_async  # type: ignore
 import google.protobuf
 
@@ -203,7 +203,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

@@ -120,51 +120,8 @@ class _Base{{ service.name }}RestTransport({{service.name}}Transport):
         def _get_unset_required_fields(cls, message_dict):
             return {k: v for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() if k not in message_dict}
         {% endif %}{# required fields #}
-
         {% set method_http_options = method.http_options %}
-
         {{ shared_macros.http_options_method(method_http_options)|indent(8) }}
-
-        @staticmethod
-        def _get_transcoded_request(http_options, request):
-            {% if method.input.ident.is_proto_plus_type %}
-            pb_request = {{method.input.ident}}.pb(request)
-            {% else %}
-            pb_request = request
-            {% endif %}
-            transcoded_request = path_template.transcode(http_options, pb_request)
-            return transcoded_request
-
-        {% set body_spec = method.http_options[0].body %}
-        {%- if body_spec %}
-
-        @staticmethod
-        def _get_request_body_json(transcoded_request):
-            # Jsonify the request body
-
-            body = json_format.MessageToJson(
-                transcoded_request['body'],
-                use_integers_for_enums={{ opts.rest_numeric_enums }}
-            )
-            return body
-
-        {%- endif %}{# body_spec #}
-
-        @staticmethod
-        def _get_query_params_json(transcoded_request):
-            query_params = json.loads(json_format.MessageToJson(
-                transcoded_request['query_params'],
-                use_integers_for_enums={{ opts.rest_numeric_enums }},
-            ))
-            {% if method.input.required_fields %}
-            query_params.update(_Base{{ service.name }}RestTransport._Base{{method.name}}._get_unset_required_fields(query_params))
-            {% endif %}{# required fields #}
-
-            {% if opts.rest_numeric_enums %}
-            query_params["$alt"] = "json;enum-encoding=int"
-            {% endif %}
-            return query_params
-
         {% endif %}{# method.http_options and not method.client_streaming #}
     {% endfor %}
 

@@ -0,0 +1,13 @@
+[run]
+branch = True
+
+[report]
+show_missing = True
+omit =
+    google/showcase/__init__.py
+    google/showcase/gapic_version.py
+exclude_lines =
+    # Re-enable the standard pragma
+    pragma: NO COVER
+    # Ignore debug-only repr
+    def __repr__
@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+# 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.
+#
+[flake8]
+# TODO(https://github.com/googleapis/gapic-generator-python/issues/2333):
+# Resolve flake8 lint issues
+ignore = E203, E231, E266, E501, W503
+exclude =
+  # TODO(https://github.com/googleapis/gapic-generator-python/issues/2333):
+  # Ensure that generated code passes flake8 lint
+  **/gapic/**
+  **/services/**
+  **/types/**
+  # Exclude Protobuf gencode
+  *_pb2.py
+
+  # Standard linting exemptions.
+  **/.nox/**
+  __pycache__,
+  .git,
+  *.pyc,
+  conf.py