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

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 Union 

26 

27import google.protobuf.message 

28 

29 

30def setup_request_id( 

31 request: Union[google.protobuf.message.Message, dict, None], 

32 field_name: str, 

33 is_proto3_optional: bool, 

34) -> None: 

35 """Populate a UUID4 field in the request if it is not already set. 

36 

37 This helper is used to ensure request idempotency by automatically 

38 generating a unique identifier (such as `request_id`) for requests 

39 that support it. If a request is retried, the same identifier can be 

40 sent on subsequent retries, allowing the server to recognize the retried 

41 request and prevent duplicate processing (e.g., creating duplicate 

42 resources). 

43 

44 Args: 

45 request (Union[google.protobuf.message.Message, dict]): The 

46 request object. 

47 field_name (str): The name of the field to populate. 

48 is_proto3_optional (bool): Whether the field is proto3 optional. 

49 """ 

50 if request is None: 

51 return 

52 

53 if isinstance(request, dict): 

54 if is_proto3_optional: 

55 if field_name not in request or request[field_name] is None: 

56 request[field_name] = str(uuid.uuid4()) 

57 elif not request.get(field_name): 

58 request[field_name] = str(uuid.uuid4()) 

59 return 

60 

61 if is_proto3_optional: 

62 try: 

63 # Pure protobuf messages 

64 if not request.HasField(field_name): 

65 setattr(request, field_name, str(uuid.uuid4())) 

66 except (AttributeError, ValueError): 

67 # Proto-plus messages or other objects 

68 if getattr(request, field_name, None) is None: 

69 setattr(request, field_name, str(uuid.uuid4())) 

70 else: 

71 if not getattr(request, field_name, None): 

72 setattr(request, field_name, str(uuid.uuid4()))