Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/rest_helpers.py: 28%

1# Copyright 2021 Google LLC 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Helpers for rest transports.""" 

16 

17import functools 

18import json 

19import operator 

20from typing import Any, Dict, List, Optional, Tuple 

21 

22from google.protobuf import json_format 

23 

24from google.api_core import path_template 

25 

26__all__ = ["flatten_query_params", "transcode", "transcode_request"] 

27 

28 

29def flatten_query_params(obj, strict=False): 

30 """Flatten a dict into a list of (name,value) tuples. 

31 

32 The result is suitable for setting query params on an http request. 

33 

34 .. code-block:: python 

35 

36 >>> obj = {'a': 

37 ... {'b': 

38 ... {'c': ['x', 'y', 'z']} }, 

39 ... 'd': 'uvw', 

40 ... 'e': True, } 

41 >>> flatten_query_params(obj, strict=True) 

42 [('a.b.c', 'x'), ('a.b.c', 'y'), ('a.b.c', 'z'), ('d', 'uvw'), ('e', 'true')] 

43 

44 Note that, as described in 

45 https://github.com/googleapis/googleapis/blob/48d9fb8c8e287c472af500221c6450ecd45d7d39/google/api/http.proto#L117, 

46 repeated fields (i.e. list-valued fields) may only contain primitive types (not lists or dicts). 

47 This is enforced in this function. 

48 

49 Args: 

50 obj: a possibly nested dictionary (from json), or None 

51 strict: a bool, defaulting to False, to enforce that all values in the 

52 result tuples be strings and, if boolean, lower-cased. 

53 

54 Returns: a list of tuples, with each tuple having a (possibly) multi-part name 

55 and a scalar value. 

56 

57 Raises: 

58 TypeError if obj is not a dict or None 

59 ValueError if obj contains a list of non-primitive values. 

60 """ 

61 

62 if obj is not None and not isinstance(obj, dict): 

63 raise TypeError("flatten_query_params must be called with dict object") 

64 

65 return _flatten(obj, key_path=[], strict=strict) 

66 

67 

68def _flatten(obj, key_path, strict=False): 

69 if obj is None: 

70 return [] 

71 if isinstance(obj, dict): 

72 return _flatten_dict(obj, key_path=key_path, strict=strict) 

73 if isinstance(obj, list): 

74 return _flatten_list(obj, key_path=key_path, strict=strict) 

75 return _flatten_value(obj, key_path=key_path, strict=strict) 

76 

77 

78def _is_primitive_value(obj): 

79 if obj is None: 

80 return False 

81 

82 if isinstance(obj, (list, dict)): 

83 raise ValueError("query params may not contain repeated dicts or lists") 

84 

85 return True 

86 

87 

88def _flatten_value(obj, key_path, strict=False): 

89 return [(".".join(key_path), _canonicalize(obj, strict=strict))] 

90 

91 

92def _flatten_dict(obj, key_path, strict=False): 

93 items = ( 

94 _flatten(value, key_path=key_path + [key], strict=strict) 

95 for key, value in obj.items() 

96 ) 

97 return functools.reduce(operator.concat, items, []) 

98 

99 

100def _flatten_list(elems, key_path, strict=False): 

101 # Only lists of scalar values are supported. 

102 # The name (key_path) is repeated for each value. 

103 items = ( 

104 _flatten_value(elem, key_path=key_path, strict=strict) 

105 for elem in elems 

106 if _is_primitive_value(elem) 

107 ) 

108 return functools.reduce(operator.concat, items, []) 

109 

110 

111def _canonicalize(obj, strict=False): 

112 if strict: 

113 value = str(obj) 

114 if isinstance(obj, bool): 

115 value = value.lower() 

116 return value 

117 return obj 

118 

119 

120def transcode_request( 

121 http_options: List[Dict[str, str]], 

122 request: Any, 

123 required_fields_default_values: Optional[Dict[str, Any]] = None, 

124 rest_numeric_enums: bool = False, 

125) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: 

126 """Transcodes a request into HTTP method, URI, body, and query parameters. 

127 

128 Args: 

129 http_options (List[Dict[str, str]]): List of HTTP transcoding rules. 

130 request (Any): The protobuf or proto-plus request message. 

131 required_fields_default_values (Optional[Dict[str, Any]]): Dictionary 

132 of required fields default values to merge into query parameters if missing. 

133 rest_numeric_enums (bool): Whether to encode enums as integers. 

134 

135 Returns: 

136 Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: 

137 - The raw transcoded request dictionary (containing keys like 'uri', 'method'). 

138 - The serialized request body JSON string, or None if no body. 

139 - The query parameters dictionary. 

140 """ 

141 if request is None: 

142 raise TypeError("request cannot be None") 

143 

144 # Convert proto-plus message to its underlying protobuf message if needed 

145 pb_request = getattr(request, "_pb", request) 

146 

147 transcoded_request = path_template.transcode(http_options, pb_request) 

148 

149 body_json = None 

150 if transcoded_request.get("body") is not None: 

151 body_json = json_format.MessageToJson( 

152 transcoded_request["body"], 

153 use_integers_for_enums=rest_numeric_enums, 

154 ) 

155 

156 query_params_json = {} 

157 if transcoded_request.get("query_params") is not None: 

158 query_params_json = json.loads( 

159 json_format.MessageToJson( 

160 transcoded_request["query_params"], 

161 use_integers_for_enums=rest_numeric_enums, 

162 ) 

163 ) 

164 

165 # If required_fields_default_values is provided, we merge default values for missing 

166 # required fields into the query parameters. 

167 if required_fields_default_values: 

168 for k, v in required_fields_default_values.items(): 

169 if k not in query_params_json: 

170 query_params_json[k] = v 

171 

172 if rest_numeric_enums: 

173 query_params_json["$alt"] = "json;enum-encoding=int" 

174 

175 return transcoded_request, body_json, query_params_json 

176 

177 

178transcode = transcode_request