diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
index ff1d3c69f91e..bba030dd0141 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
@@ -12,4 +12,53 @@ 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. #}
 
+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()))
 {% endblock %}
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
index 755e4530e7ba..5f1b236f1780 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
@@ -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 #}
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
index 5fe416902b86..7d9809bcb02c 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
@@ -8,9 +8,6 @@ import logging as std_logging
 from collections import OrderedDict
 import re
 from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}AsyncIterable, Awaitable, {% endif %}{% if service.any_client_streaming %}AsyncIterator, {% endif %}Sequence, Tuple, Type, Union
-{% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
-import uuid
-{% endif %}
 {% if service.any_deprecated %}
 import warnings
 {% endif %}
@@ -21,6 +18,16 @@ 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
+{% set has_auto_populated_fields = namespace(value=false) %}
+{% for method in service.methods.values() %}
+  {% set method_settings = api.all_method_settings.get(method.meta.address.proto) %}
+  {% if method_settings and method_settings.auto_populated_fields %}
+    {% set has_auto_populated_fields.value = true %}
+  {% endif %}
+{% endfor %}
+{% if has_auto_populated_fields.value %}
+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
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
index e2e3edb24967..451d28e9ea32 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
@@ -16,9 +16,6 @@ import logging as std_logging
 import os
 import re
 from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}Iterable, {% endif %}{% if service.any_client_streaming %}Iterator, {% endif %}Sequence, Tuple, Type, Union, cast
-{% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
-import uuid
-{% endif %}
 import warnings
 
 {% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
@@ -30,6 +27,16 @@ from google.api_core import exceptions as core_exceptions
 from google.api_core import extended_operation
 {% endif %}
 from google.api_core import gapic_v1
+{% set has_auto_populated_fields = namespace(value=false) %}
+{% for method in service.methods.values() %}
+  {% set method_settings = api.all_method_settings.get(method.meta.address.proto) %}
+  {% if method_settings and method_settings.auto_populated_fields %}
+    {% set has_auto_populated_fields.value = true %}
+  {% endif %}
+{% endfor %}
+{% if has_auto_populated_fields.value %}
+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,37 +460,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         # NOTE (b/349488459): universe validation is disabled until further notice.
         return True
 
-    {% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
-    @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,
@@ -588,12 +565,13 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index c432b75621ea..2d19415208a4 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -6,9 +6,17 @@
 {% import "tests/unit/gapic/%name_%version/%sub/test_macros.j2" as test_macros %}
 {% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
+{% set has_auto_populated_fields = namespace(value=false) %}
+{% for method in service.methods.values() %}
+  {% set method_settings = api.all_method_settings.get(method.meta.address.proto) %}
+  {% if method_settings and method_settings.auto_populated_fields %}
+    {% set has_auto_populated_fields.value = true %}
+  {% endif %}
+{% endfor %}
+
 import os
 import asyncio
-{% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
+{% if has_auto_populated_fields.value %}
 import re
 {% endif %}
 from unittest import mock
@@ -107,7 +115,7 @@ CRED_INFO_JSON = {
     "principal": "service-account@example.com",
 }
 CRED_INFO_STRING = json.dumps(CRED_INFO_JSON)
-{% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
+{% if has_auto_populated_fields.value %}
 _UUID4_RE = re.compile(r"{{ uuid4_re }}")
 {% endif %}
 
@@ -434,84 +442,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 api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|select|list %}
-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"),
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2
index e221d512a3e8..a7bf78e92587 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_compat.py.j2
@@ -9,4 +9,91 @@ 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. #}
 
+
+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
+
 {% endblock %}
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
index bccc38afe2a1..982b2fb19b44 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
@@ -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 #}
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
index 590b6fa1c615..34179117a46e 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
@@ -611,12 +611,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
index 1a367689d14e..033ffc3f728a 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_compat.py
index 8491204d1041..8d161872be3f 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.asset_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
index 4ad970da32e9..4417d8b567a8 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
@@ -548,12 +548,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_compat.py
index 8491204d1041..e96025be2917 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.iam.credentials_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
index 508e17a7e889..e8aa3a66ea36 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
index 7255d3c91709..859d9a0d3065 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
@@ -731,12 +731,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_compat.py
index 8491204d1041..819f64e06527 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.eventarc_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
index b9495ef6e42d..abf72a70c84b 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
index 15922e6d865a..4c2cfdd11512 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
@@ -604,12 +604,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
index e89762755eda..38fb5c326a03 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
@@ -535,12 +535,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
index 90e9355f8c26..d78aa2465c12 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
@@ -536,12 +536,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_compat.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_compat.py
index 8491204d1041..ae3114fbc5ca 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.logging_v2._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
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
index 8cc17d810664..2fc7b42dd627 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
index fd56f0210d78..d91208a0f85b 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
index 762b4b3ab94d..66b33dda85b2 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
index 61204cb87a52..f076e428a5d0 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
@@ -604,12 +604,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
index e89762755eda..38fb5c326a03 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
@@ -535,12 +535,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
index fa55137223d1..fd9206a74471 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
@@ -536,12 +536,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_compat.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_compat.py
index 8491204d1041..ae3114fbc5ca 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.logging_v2._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
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
index 93731051d1bc..eff6451871eb 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
index fd56f0210d78..d91208a0f85b 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
index 090ea4a91bec..ce9fa5bcde77 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
index 33ccce478c8e..2132cb798f33 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
@@ -576,12 +576,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
index d58d99d1fdb8..f7dc40c32a46 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_compat.py
index 8491204d1041..29d9c09360e7 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.redis_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
index 031573ef83d7..e74cce01c8c8 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
@@ -576,12 +576,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 780964608350..45f5c691b219 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 from unittest import mock
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_compat.py
index 8491204d1041..29d9c09360e7 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.redis_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
index 7c684e94b03d..1b76bd6ca40b 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
@@ -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()))
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/async_client.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/async_client.py
index 40eaca9be991..04f8e1ba4008 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/async_client.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/async_client.py
@@ -17,13 +17,13 @@
 from collections import OrderedDict
 import re
 from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, Sequence, Tuple, Type, Union
-import uuid
 
 from google.cloud.storagebatchoperations_v1 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
+from google.cloud.storagebatchoperations_v1._compat import setup_request_id
 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
@@ -621,7 +621,7 @@ async def sample_create_job():
             )),
         )
 
-        self._client._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._client._validate_universe_domain()
@@ -727,7 +727,7 @@ async def sample_delete_job():
             )),
         )
 
-        self._client._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._client._validate_universe_domain()
@@ -829,7 +829,7 @@ async def sample_cancel_job():
             )),
         )
 
-        self._client._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._client._validate_universe_domain()
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
index 5f79cf8e016a..b02e74e0c2e4 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
@@ -20,7 +20,6 @@
 import os
 import re
 from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, Sequence, Tuple, Type, Union, cast
-import uuid
 import warnings
 
 from google.cloud.storagebatchoperations_v1 import gapic_version as package_version
@@ -28,6 +27,7 @@
 from google.api_core import client_options as client_options_lib
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
+from google.cloud.storagebatchoperations_v1._compat import setup_request_id
 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
@@ -471,36 +471,6 @@ def _validate_universe_domain(self):
         # NOTE (b/349488459): universe validation is disabled until further notice.
         return True
 
-    @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()))
-
     def _add_cred_info_for_auth_errors(
         self,
         error: core_exceptions.GoogleAPICallError
@@ -601,12 +571,13 @@ def __init__(self, *,
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        self._client_options = client_options
-        if isinstance(self._client_options, dict):
-            self._client_options = client_options_lib.from_dict(self._client_options)
-        if self._client_options is None:
-            self._client_options = client_options_lib.ClientOptions()
-        self._client_options = cast(client_options_lib.ClientOptions, self._client_options)
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+        self._client_options: client_options_lib.ClientOptions = cast(
+            client_options_lib.ClientOptions, client_options
+        )
 
         universe_domain_opt = getattr(self._client_options, 'universe_domain', None)
 
@@ -1035,7 +1006,7 @@ def sample_create_job():
             )),
         )
 
-        self._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._validate_universe_domain()
@@ -1140,7 +1111,7 @@ def sample_delete_job():
             )),
         )
 
-        self._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._validate_universe_domain()
@@ -1241,7 +1212,7 @@ def sample_cancel_job():
             )),
         )
 
-        self._setup_request_id(request, 'request_id', False)
+        setup_request_id(request, 'request_id', False)
 
         # Validate the universe domain.
         self._validate_universe_domain()
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_compat.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_compat.py
index 8491204d1041..d84f0d95b840 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_compat.py
@@ -14,3 +14,89 @@
 # limitations under the License.
 #
 """Tests for the compatibility module for older versions of google-api-core."""
+
+
+import re
+import pytest
+
+from google.cloud.storagebatchoperations_v1._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
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py
index 892375775385..6062293b6072 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
 import os
 import asyncio
 import re
@@ -351,82 +352,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 == []
 
-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()
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", True)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request.request_id)
-
-    # Test with proto3 optional field already in request
-    request = MockRequest(request_id="already_set")
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", True)
-    assert request.request_id == "already_set"
-
-    # Test with non-proto3 optional field empty
-    request = MockRequest(request_id="")
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", False)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request.request_id)
-
-    # Test with non-proto3 optional field already set
-    request = MockRequest(request_id="already_set")
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", False)
-    assert request.request_id == "already_set"
-
-    # Test with proto3 optional field not in request (MockProtoRequest)
-    request = MockProtoRequest()
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", True)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request.request_id)
-
-    # Test with proto3 optional field already in request (MockProtoRequest)
-    request = MockProtoRequest(request_id="already_set")
-    StorageBatchOperationsClient._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()
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", True)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request.request_id)
-
-    # Test with dict and proto3 optional field not in request
-    request = {}
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", True)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request["request_id"])
-
-    # Test with dict and proto3 optional field already in request
-    request = {"request_id": "already_set"}
-    StorageBatchOperationsClient._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": ""}
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", False)
-    assert re.match(r"[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}", request["request_id"])
-
-    # Test with dict and non-proto3 optional field already set
-    request = {"request_id": "already_set"}
-    StorageBatchOperationsClient._setup_request_id(request, "request_id", False)
-    assert request["request_id"] == "already_set"
-
 @pytest.mark.parametrize("client_class,transport_name", [
     (StorageBatchOperationsClient, "grpc"),
     (StorageBatchOperationsAsyncClient, "grpc_asyncio"),
diff --git a/packages/gapic-generator/tests/unit/generator/test_generator.py b/packages/gapic-generator/tests/unit/generator/test_generator.py
index 9d8545c4192f..519a2b94b113 100644
--- a/packages/gapic-generator/tests/unit/generator/test_generator.py
+++ b/packages/gapic-generator/tests/unit/generator/test_generator.py
@@ -136,6 +136,30 @@ def test_get_response_ignores_private_files():
             assert cgr.file[0].content == "I am a template result.\n"
 
 
+def test_get_response_renders_allowed_private_templates():
+    generator_obj = make_generator()
+    with mock.patch.object(jinja2.FileSystemLoader, "list_templates") as list_templates:
+        list_templates.return_value = [
+            "foo/bar/__init__.py.j2",
+            "foo/bar/_compat.py.j2",
+            "foo/bar/_ignored.py.j2",
+        ]
+        with mock.patch.object(jinja2.Environment, "get_template") as get_template:
+            get_template.return_value = jinja2.Template("I am a template result.")
+            cgr = generator_obj.get_response(
+                api_schema=make_api(), opts=Options.build("")
+            )
+            list_templates.assert_called_once()
+            get_template.assert_has_calls(
+                [
+                    mock.call("foo/bar/__init__.py.j2"),
+                    mock.call("foo/bar/_compat.py.j2"),
+                ],
+                any_order=True,
+            )
+            assert len(cgr.file) == 2
+
+
 def test_get_response_fails_invalid_file_paths():
     generator_obj = make_generator()
     with mock.patch.object(jinja2.FileSystemLoader, "list_templates") as list_templates: