1# -*- coding: utf-8 -*-
2# Copyright 2026 Google LLC
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17"""Helpers for preparing and structuring API requests.
18
19This module provides utilities to preprocess request parameters and objects
20before invoking API methods, such as automatically generating request IDs
21if they are not already set.
22"""
23
24import uuid
25from typing import TYPE_CHECKING, Union
26
27import google.protobuf.message
28
29if TYPE_CHECKING: # pragma: NO COVER
30 import proto # type: ignore[import-untyped]
31
32
33def setup_request_id(
34 request: Union[google.protobuf.message.Message, "proto.Message", dict, None],
35 field_name: str,
36 is_proto3_optional: bool,
37) -> None:
38 """Populate a UUID4 field in the request if it is not already set.
39
40 This helper is used to ensure request idempotency by automatically
41 generating a unique identifier (such as `request_id`) for requests
42 that support it. If a request is retried, the same identifier can be
43 sent on subsequent retries, allowing the server to recognize the retried
44 request and prevent duplicate processing (e.g., creating duplicate
45 resources).
46
47 Args:
48 request (Union[google.protobuf.message.Message, proto.Message, dict, None]): The
49 request object or dictionary.
50 field_name (str): The name of the field to populate (e.g., "request_id").
51 is_proto3_optional (bool): Whether the field supports explicit presence
52 (defined with `optional` in proto3 syntax). When True, empty strings ("")
53 are preserved as explicit user input per AIP-4235, and UUID auto-population
54 occurs only if the field is unset. When False, any empty or falsy value is
55 populated with a UUID.
56 """
57 if request is None:
58 return
59
60 # Evaluate whether the field is considered "unset" and needs auto-population.
61 #
62 # According to AIP-4235, optional request ID fields must be populated
63 # if and only if they have explicit presence (`is_proto3_optional=True`)
64 # and were not set by the user (i.e. unset). Explicitly provided empty
65 # strings ('') must be preserved when `is_proto3_optional=True`.
66 should_populate = False
67 if isinstance(request, dict):
68 if is_proto3_optional:
69 # Case 1a: Dictionary request with explicit presence (`is_proto3_optional=True`).
70 # Per AIP-4235, auto-populate only if the key is completely missing from
71 # the dictionary or its value is explicitly set to None.
72 # An explicit empty string ('') must NOT be overwritten.
73 should_populate = field_name not in request or request[field_name] is None
74 else:
75 # Case 1b: Dictionary request without explicit presence (`is_proto3_optional=False`).
76 # Auto-populate if the key is missing, None, or falsy (e.g., empty string '').
77 should_populate = not request.get(field_name)
78 else:
79 if is_proto3_optional:
80 # Case 2a: Proto request with explicit presence (`is_proto3_optional=True`)
81 # (proto-plus wrapper or pure protobuf message).
82 # Extract the protobuf from proto-plus if wrapped.
83 pure_pb: google.protobuf.message.Message = getattr(request, "_pb", request)
84 try:
85 should_populate = not pure_pb.HasField(field_name)
86 except (AttributeError, ValueError):
87 # Fall back if `HasField` fails or is unsupported.
88 should_populate = getattr(pure_pb, field_name, None) is None
89 else:
90 # Case 2b: Proto request without explicit presence (`is_proto3_optional=False`).
91 # Auto-populate if the field value is falsy (None or empty string '').
92 should_populate = not bool(getattr(request, field_name, False))
93
94 # If the field was found to be empty, set random id
95 if should_populate:
96 generated_id = str(uuid.uuid4())
97 if isinstance(request, dict):
98 request[field_name] = generated_id
99 else:
100 setattr(request, field_name, generated_id)