Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/gapic_v1/requests.py: 20%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

25 statements  

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)