feat(api-core): centralize rest transcoding helpers

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

@@ -16,6 +16,12 @@
 
 import functools
 import operator
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+from google.api_core import path_template
+from google.protobuf import json_format
+
+__all__ = ["flatten_query_params", "transcode", "transcode_request"]
 
 
 def flatten_query_params(obj, strict=False):
@@ -107,3 +113,88 @@ def _canonicalize(obj, strict=False):
             value = value.lower()
         return value
     return obj
+
+
+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_format.MessageToDict(
+            transcoded_request["query_params"],
+            use_integers_for_enums=rest_numeric_enums,
+        )
+
+    if required_fields_default_values:
+        matched_option = None
+        for option in http_options:
+            if (
+                option.get("method", "").lower()
+                == transcoded_request.get("method", "").lower()
+            ):
+                if path_template.validate(
+                    option.get("uri", ""), transcoded_request.get("uri", "")
+                ):
+                    matched_option = option
+                    break
+
+        bound_fields: Optional[Set[str]] = set()
+        if matched_option:
+            body_param = matched_option.get("body")
+            if body_param == "*":
+                bound_fields = None
+            else:
+                assert bound_fields is not None
+                uri_template = matched_option.get("uri", "")
+                for m in path_template._VARIABLE_RE.finditer(uri_template):
+                    bound_fields.add(m.group("name").split(".")[0])
+                if body_param:
+                    bound_fields.add(body_param.split(".")[0])
+
+        if bound_fields is not None:
+            for k, v in required_fields_default_values.items():
+                if k in bound_fields:
+                    continue
+                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
+
+
+transcode = transcode_request

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