feat(api-core): centralize rest transcoding helpers

packages/google-api-core/google/api_core/rest_helpers.py

@@ -16,6 +16,11 @@
 
 import functools
 import operator
+from typing import Any, Dict, List, Optional, Tuple
+
+from google.protobuf import json_format
+
+from google.api_core import path_template
 
 
 def flatten_query_params(obj, strict=False):
@@ -107,3 +112,54 @@ def _canonicalize(obj, strict=False):
             value = value.lower()
         return value
     return obj
+
+
+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]]:
+    """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.
+    """
+    # 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_format.MessageToDict(
+            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

packages/google-api-core/tests/conftest.py

@@ -0,0 +1,31 @@
+# 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.
+
+import os
+from unittest import mock
+
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def mock_mtls_env():
+    """Autouse session-scoped fixture to isolate unit tests from workstation mTLS environments."""
+    with mock.patch.dict(
+        os.environ,
+        {
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false",
+            "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false",
+        },
+    ):
+        yield

packages/google-api-core/tests/unit/test_rest_helpers.py

@@ -12,9 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import pytest
+from google.protobuf import descriptor_pb2
 
 from google.api_core import rest_helpers
+from google.api_core.rest_helpers import transcode
 
 
 def test_flatten_simple_value():
@@ -92,3 +95,120 @@ def test_flatten_repeated_list():
 
     with pytest.raises(ValueError):
         rest_helpers.flatten_query_params(obj)
+
+
+def test_transcode_basic():
+    # We use FieldDescriptorProto as it has standard primitive fields and nested messages.
+    http_options = [
+        {
+            "method": "get",
+            "uri": "/v1/test/{name}",
+        }
+    ]
+
+    request = descriptor_pb2.FieldDescriptorProto()
+    request.name = "my-field"
+    request.number = 123
+
+    transcoded, body, query_params = transcode(http_options, request)
+
+    assert transcoded["method"] == "get"
+    assert transcoded["uri"] == "/v1/test/my-field"
+    assert body is None
+    # 'number' should be in query parameters
+    assert "number" in query_params
+    assert query_params["number"] == 123
+
+
+def test_transcode_with_nested_field():
+    http_options = [
+        {
+            "method": "get",
+            "uri": "/v1/test/{options.deprecated}/{name}",
+        }
+    ]
+
+    request = descriptor_pb2.FieldDescriptorProto()
+    request.name = "my-field"
+    request.options.deprecated = True
+    request.number = 123
+
+    transcoded, body, query_params = transcode(http_options, request)
+
+    assert transcoded["method"] == "get"
+    assert transcoded["uri"] == "/v1/test/True/my-field"
+    assert body is None
+    assert "number" in query_params
+    assert query_params["number"] == 123
+
+
+def test_transcode_with_body():
+    http_options = [
+        {
+            "method": "post",
+            "uri": "/v1/test/{name}",
+            "body": "options",
+        }
+    ]
+
+    request = descriptor_pb2.FieldDescriptorProto()
+    request.name = "my-field"
+    request.options.deprecated = True
+    request.number = 123
+
+    transcoded, body, query_params = transcode(http_options, request)
+
+    assert transcoded["method"] == "post"
+    assert transcoded["uri"] == "/v1/test/my-field"
+    assert body is not None
+    body_data = json.loads(body)
+    assert body_data["deprecated"] is True
+    # Query parameters should not contain 'options' (the body)
+    assert "number" in query_params
+    assert query_params["number"] == 123
+    assert "options" not in query_params
+
+
+def test_transcode_with_required_fields_default_values():
+    http_options = [
+        {
+            "method": "get",
+            "uri": "/v1/test/{name}",
+        }
+    ]
+
+    request = descriptor_pb2.FieldDescriptorProto()
+    request.name = "my-field"
+
+    required_defaults = {"requiredQueryParam": "default-val"}
+
+    transcoded, body, query_params = transcode(
+        http_options,
+        request,
+        required_fields_default_values=required_defaults,
+    )
+
+    assert query_params["requiredQueryParam"] == "default-val"
+
+
+def test_transcode_with_numeric_enums():
+    http_options = [
+        {
+            "method": "get",
+            "uri": "/v1/test/{name}",
+        }
+    ]
+
+    request = descriptor_pb2.FieldDescriptorProto()
+    request.name = "my-field"
+    request.type = descriptor_pb2.FieldDescriptorProto.TYPE_STRING
+
+    # Without numeric enums
+    _, _, query_params = transcode(http_options, request, rest_numeric_enums=False)
+    assert query_params["type"] == "TYPE_STRING"
+
+    # With numeric enums
+    _, _, query_params = transcode(http_options, request, rest_numeric_enums=True)
+    # Type number for TYPE_STRING is 9
+    assert query_params["type"] == 9
+    assert query_params["$alt"] == "json;enum-encoding=int"