googleapis · hebaalazzeh · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026
@@ -13,6 +13,13 @@ falling back to the local implementation if not present. #}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17883): 
 Backfill compatibility functions being removed from the client layer. #}
 
+import json
+
+from typing import Any, Dict, List, Optional, Tuple
+
+from google.api_core import path_template
+from google.protobuf import json_format
+
 {% if has_auto_populated_fields %}
 from typing import Union
 import uuid
@@ -64,4 +71,62 @@ def setup_request_id(
         if not getattr(request, field_name, None):
             setattr(request, field_name, str(uuid.uuid4()))
 {% endif %}
+
+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,
+            )
+        )
+
+    # 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"
+
+    return transcoded_request, body_json, query_params_json
+
 {% endblock %}
@@ -194,22 +194,24 @@ 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 #} 
+    {# 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 = transcode_request(
+        http_options,
+        request,
+        required_fields_default_values=getattr(
+            _Base{{ service_name }}RestTransport._Base{{method_name}},
+            "_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,5 @@
 
         {{ 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
-
     {% endfor %}
     {% endif %} {# rest in opts.transport #}
@@ -13,6 +13,8 @@ from google.api_core import retry as retries
 from google.api_core import rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}}._compat import transcode_request
 import google.protobuf
 
 from google.protobuf import json_format
@@ -245,7 +247,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

@@ -34,6 +34,9 @@ from google.cloud.location import locations_pb2 # type: ignore
 from google.api_core import retry_async as retries
 from google.api_core import rest_helpers
 from google.api_core import rest_streaming_async  # type: ignore
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}}._compat import transcode_request
+
 import google.protobuf
 
 from google.protobuf import json_format
@@ -203,7 +206,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,10 @@ 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
 
+        {{ shared_macros.http_options_method(method_http_options)|indent(8) }}
         {% endif %}{# method.http_options and not method.client_streaming #}
     {% endfor %}