googleapis · hebaalazzeh · Jul 27, 2026 · Jul 16, 2026 · Jul 16, 2026 · Jul 16, 2026
@@ -3,6 +3,7 @@
 {% block content %}
 
 """A compatibility module for older versions of google-api-core."""
+{% set has_auto_populated_fields = api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17813):
 Clean up this file/functions when the minimum supported version of 
 google-api-core has the functions in `_compat.py.j2`. #}
@@ -12,4 +13,55 @@ 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. #}
 
+{% if has_auto_populated_fields %}
+from typing import Union
+import uuid
+
+import google.protobuf.message
+
+
+def setup_request_id(
+ request: Union[google.protobuf.message.Message, dict, None],
+ field_name: str,
+ is_proto3_optional: bool,
+) -> None:
+ """Populate a UUID4 field in the request if it is not already set.
+
+ This helper is used to ensure request idempotency by automatically
+ generating a unique identifier (such as `request_id`) for requests
+ that support it. If a request is retried, the same identifier can be
+ sent on subsequent retries, allowing the server to recognize the retried
+ request and prevent duplicate processing (e.g., creating duplicate
+ resources).
+
+ 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.
+ """
+ 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] = str(uuid.uuid4())
+ elif not request.get(field_name):
+ request[field_name] = str(uuid.uuid4())
+ return
+
+ if is_proto3_optional:
+ try:
+ # Pure protobuf messages
+ if not request.HasField(field_name):
+ setattr(request, field_name, str(uuid.uuid4()))
+ except (AttributeError, ValueError):
+ # Proto-plus messages or other objects
+ if not getattr(request, field_name, None):
+ setattr(request, field_name, str(uuid.uuid4()))
+ else:
+ if not getattr(request, field_name, None):
+ setattr(request, field_name, str(uuid.uuid4()))
+{% endif %}
 {% endblock %}
@@ -28,11 +28,7 @@
 {% if method_settings is not none %}
 {% for auto_populated_field in method_settings.auto_populated_fields %}
  {% set is_proto3_optional = method.input.fields[auto_populated_field].proto3_optional %}
- {% if is_async %}
- self._client._setup_request_id(request, '{{ auto_populated_field }}', {{ is_proto3_optional }})
- {% else %}
- self._setup_request_id(request, '{{ auto_populated_field }}', {{ is_proto3_optional }})
- {% endif %}
+ setup_request_id(request, '{{ auto_populated_field }}', {{ is_proto3_optional }})
 {% endfor %}
 {% endif %}{# if method_settings is not none #}
 {% endwith %}{# method_settings #}

@@ -21,6 +21,9 @@ from {{package_path}} import gapic_version as package_version
 from google.api_core.client_options import ClientOptions
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+{% if has_auto_populated_fields %}
+from {{package_path}}._compat import setup_request_id
+{% endif %}
 from google.api_core import retry_async as retries
 from google.auth import credentials as ga_credentials # type: ignore
 from google.oauth2 import service_account # type: ignore

@@ -30,6 +30,9 @@ from google.api_core import exceptions as core_exceptions
 from google.api_core import extended_operation
 {% endif %}
 from google.api_core import gapic_v1
+{% if has_auto_populated_fields %}
+from {{package_path}}._compat import setup_request_id
+{% endif %}
 from google.api_core import retry as retries
 from google.auth import credentials as ga_credentials # type: ignore
 from google.auth.transport import mtls # type: ignore
@@ -453,39 +456,6 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
  # NOTE (b/349488459): universe validation is disabled until further notice.
  return True
 
- {% if has_auto_populated_fields %}
-
- @staticmethod
- 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.
- """
- if isinstance(request, dict):
- if is_proto3_optional:
- if field_name not in request:
- request[field_name] = str(uuid.uuid4())
- elif not request.get(field_name):
- request[field_name] = str(uuid.uuid4())
- return
-
- if is_proto3_optional:
- try:
- # Pure protobuf messages
- if not request.HasField(field_name):
- setattr(request, field_name, str(uuid.uuid4()))
- except (AttributeError, ValueError):
- # Proto-plus messages or other objects
- if field_name not in request:
- setattr(request, field_name, str(uuid.uuid4()))
- else:
- if not getattr(request, field_name):
- setattr(request, field_name, str(uuid.uuid4()))
- {% endif %}
-
  def _add_cred_info_for_auth_errors(
  self,
  error: core_exceptions.GoogleAPICallError

@@ -434,84 +434,6 @@ def test__add_cred_info_for_auth_errors_no_get_cred_info(error_code):
  client._add_cred_info_for_auth_errors(error)
  assert error.details == []
 
-{% if has_auto_populated_fields %}
-def test__setup_request_id():
- class MockRequest:
- def __init__(self, **kwargs):
- for k, v in kwargs.items():
- setattr(self, k, v)
- def __contains__(self, key):
- return hasattr(self, key)
-
- class MockProtoRequest:
- def __init__(self, **kwargs):
- for k, v in kwargs.items():
- setattr(self, k, v)
- def HasField(self, key):
- return hasattr(self, key)
-
- # Test with proto3 optional field not in request
- request = MockRequest()
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request.request_id)
-
- # Test with proto3 optional field already in request
- request = MockRequest(request_id="already_set")
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert request.request_id == "already_set"
-
- # Test with non-proto3 optional field empty
- request = MockRequest(request_id="")
- {{ service.client_name }}._setup_request_id(request, "request_id", False)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request.request_id)
-
- # Test with non-proto3 optional field already set
- request = MockRequest(request_id="already_set")
- {{ service.client_name }}._setup_request_id(request, "request_id", False)
- assert request.request_id == "already_set"
-
- # Test with proto3 optional field not in request (MockProtoRequest)
- request = MockProtoRequest()
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request.request_id)
-
- # Test with proto3 optional field already in request (MockProtoRequest)
- request = MockProtoRequest(request_id="already_set")
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert request.request_id == "already_set"
-
- # Test with ValueError
- class MockValueErrorRequest:
- def HasField(self, key):
- raise ValueError("Mismatched field")
- def __contains__(self, key):
- return hasattr(self, key)
-
- request = MockValueErrorRequest()
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request.request_id)
-
- # Test with dict and proto3 optional field not in request
- request = {}
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request["request_id"])
-
- # Test with dict and proto3 optional field already in request
- request = {"request_id": "already_set"}
- {{ service.client_name }}._setup_request_id(request, "request_id", True)
- assert request["request_id"] == "already_set"
-
- # Test with dict and non-proto3 optional field empty
- request = {"request_id": ""}
- {{ service.client_name }}._setup_request_id(request, "request_id", False)
- assert re.match(r"{{ test_macros.get_uuid4_re() }}", request["request_id"])
-
- # Test with dict and non-proto3 optional field already set
- request = {"request_id": "already_set"}
- {{ service.client_name }}._setup_request_id(request, "request_id", False)
- assert request["request_id"] == "already_set"
-
-{% endif %}
 @pytest.mark.parametrize("client_class,transport_name", [
  {% if 'grpc' in opts.transport %}
  ({{ service.client_name }}, "grpc"),

@@ -2,11 +2,101 @@
 
 {% block content %}
 
+{% set has_auto_populated_fields = api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
 """Tests for the compatibility module for older versions of google-api-core."""
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17813):
 Clean up this file/tests when the minimum supported version of 
 google-api-core has the functions in `_compat.py.j2`. #}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17883): 
 Backfill compatibility functions tests being removed from the client layer. #}
 
+
+{% if has_auto_populated_fields %}
+import re
+import pytest
+
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}}._compat import setup_request_id
+
+class MockRequest:
+ def __init__(self, **kwargs):
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
+ def __contains__(self, key):
+ return hasattr(self, key)
+
+
+class MockProtoRequest:
+ def __init__(self, **kwargs):
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
+ def HasField(self, key):
+ return hasattr(self, key)
+
+
+class MockValueErrorRequest:
+ def HasField(self, key):
+ raise ValueError("Mismatched field")
+
+ def __contains__(self, key):
+ return hasattr(self, key)
+
+UUID_REGEX = r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"
+
+@pytest.mark.parametrize(
+ "request_obj, is_proto3_optional, expected",
+ [
+ (MockRequest(), True, "uuid"),
+ (MockRequest(request_id="already_set"), True, "already_set"),
+ (MockRequest(request_id=""), False, "uuid"),
+ (MockRequest(request_id="already_set"), False, "already_set"),
+ (MockProtoRequest(), True, "uuid"),
+ (MockProtoRequest(request_id="already_set"), True, "already_set"),
+ (MockValueErrorRequest(), True, "uuid"),
+ ({}, True, "uuid"),
+ ({"request_id": None}, True, "uuid"),
+ ({"request_id": "already_set"}, True, "already_set"),
+ ({"request_id": ""}, False, "uuid"),
+ ({"request_id": None}, False, "uuid"),
+ ({"request_id": "already_set"}, False, "already_set"),
+ (None, True, "none"),
+ ],
+ ids=[
+ "proto3_optional_not_in_request",
+ "proto3_optional_already_in_request",
+ "non_proto3_optional_empty",
+ "non_proto3_optional_already_set",
+ "proto3_optional_not_in_request_proto",
+ "proto3_optional_already_in_request_proto",
+ "value_error_fallback",
+ "dict_proto3_optional_not_in_request",
+ "dict_proto3_optional_value_none",
+ "dict_proto3_optional_already_in_request",
+ "dict_non_proto3_optional_empty",
+ "dict_non_proto3_optional_value_none",
+ "dict_non_proto3_optional_already_set",
+ "none_request",
+ ],
+)
+def test_setup_request_id(request_obj, is_proto3_optional, expected):
+ setup_request_id(request_obj, "request_id", is_proto3_optional)
+
+ if expected == "none":
+ assert request_obj is None
+ return
+
+ value = (
+ request_obj["request_id"]
+ if isinstance(request_obj, dict)
+ else request_obj.request_id
+ )
+
+ if expected == "uuid":
+ assert re.match(UUID_REGEX, value)
+ else:
+ assert value == expected
+
+{% endif %}
 {% endblock %}
@@ -2215,10 +2215,6 @@ def test_initialize_client_w_{{transport_name}}():
 {% endfor %}{# method in service.methods.values() #}
 {% endmacro %}{# empty_call_test #}
 
-{% macro get_uuid4_re() -%}
-{{ uuid4_re }}
-{%- endmacro %}{# uuid_re #}
-
 {% macro routing_parameter_test(service, api, transport, is_async) %}
 {% for method in service.methods.values() %}{# method #}
 {# See existing proposal b/330610501 to add support for explicit routing in BIDI/client side streaming #}

@@ -14,3 +14,53 @@
 # limitations under the License.
 #
 """A compatibility module for older versions of google-api-core."""
+
+from typing import Union
+import uuid
+
+import google.protobuf.message
+
+
+def setup_request_id(
+ request: Union[google.protobuf.message.Message, dict, None],
+ field_name: str,
+ is_proto3_optional: bool,
+) -> None:
+ """Populate a UUID4 field in the request if it is not already set.
+
+ This helper is used to ensure request idempotency by automatically
+ generating a unique identifier (such as `request_id`) for requests
+ that support it. If a request is retried, the same identifier can be
+ sent on subsequent retries, allowing the server to recognize the retried
+ request and prevent duplicate processing (e.g., creating duplicate
+ resources).
+
+ 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.
+ """
+ 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] = str(uuid.uuid4())
+ elif not request.get(field_name):
+ request[field_name] = str(uuid.uuid4())
+ return
+
+ if is_proto3_optional:
+ try:
+ # Pure protobuf messages
+ if not request.HasField(field_name):
+ setattr(request, field_name, str(uuid.uuid4()))
+ except (AttributeError, ValueError):
+ # Proto-plus messages or other objects
+ if not getattr(request, field_name, None):
+ setattr(request, field_name, str(uuid.uuid4()))
+ else:
+ if not getattr(request, field_name, None):
+ setattr(request, field_name, str(uuid.uuid4()))