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
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.
15"""Helpers for rest transports."""
17import functools
18import json
19import operator
20from typing import Any, Dict, List, Optional, Tuple
22from google.api_core import path_template
23from google.protobuf import json_format
25__all__ = ["flatten_query_params", "transcode", "transcode_request"]
28def flatten_query_params(obj, strict=False):
29 """Flatten a dict into a list of (name,value) tuples.
31 The result is suitable for setting query params on an http request.
33 .. code-block:: python
35 >>> obj = {'a':
36 ... {'b':
37 ... {'c': ['x', 'y', 'z']} },
38 ... 'd': 'uvw',
39 ... 'e': True, }
40 >>> flatten_query_params(obj, strict=True)
41 [('a.b.c', 'x'), ('a.b.c', 'y'), ('a.b.c', 'z'), ('d', 'uvw'), ('e', 'true')]
43 Note that, as described in
44 https://github.com/googleapis/googleapis/blob/48d9fb8c8e287c472af500221c6450ecd45d7d39/google/api/http.proto#L117,
45 repeated fields (i.e. list-valued fields) may only contain primitive types (not lists or dicts).
46 This is enforced in this function.
48 Args:
49 obj: a possibly nested dictionary (from json), or None
50 strict: a bool, defaulting to False, to enforce that all values in the
51 result tuples be strings and, if boolean, lower-cased.
53 Returns: a list of tuples, with each tuple having a (possibly) multi-part name
54 and a scalar value.
56 Raises:
57 TypeError if obj is not a dict or None
58 ValueError if obj contains a list of non-primitive values.
59 """
61 if obj is not None and not isinstance(obj, dict):
62 raise TypeError("flatten_query_params must be called with dict object")
64 return _flatten(obj, key_path=[], strict=strict)
67def _flatten(obj, key_path, strict=False):
68 if obj is None:
69 return []
70 if isinstance(obj, dict):
71 return _flatten_dict(obj, key_path=key_path, strict=strict)
72 if isinstance(obj, list):
73 return _flatten_list(obj, key_path=key_path, strict=strict)
74 return _flatten_value(obj, key_path=key_path, strict=strict)
77def _is_primitive_value(obj):
78 if obj is None:
79 return False
81 if isinstance(obj, (list, dict)):
82 raise ValueError("query params may not contain repeated dicts or lists")
84 return True
87def _flatten_value(obj, key_path, strict=False):
88 return [(".".join(key_path), _canonicalize(obj, strict=strict))]
91def _flatten_dict(obj, key_path, strict=False):
92 items = (
93 _flatten(value, key_path=key_path + [key], strict=strict)
94 for key, value in obj.items()
95 )
96 return functools.reduce(operator.concat, items, [])
99def _flatten_list(elems, key_path, strict=False):
100 # Only lists of scalar values are supported.
101 # The name (key_path) is repeated for each value.
102 items = (
103 _flatten_value(elem, key_path=key_path, strict=strict)
104 for elem in elems
105 if _is_primitive_value(elem)
106 )
107 return functools.reduce(operator.concat, items, [])
110def _canonicalize(obj, strict=False):
111 if strict:
112 value = str(obj)
113 if isinstance(obj, bool):
114 value = value.lower()
115 return value
116 return obj
119def transcode_request(
120 http_options: List[Dict[str, str]],
121 request: Any,
122 required_fields_default_values: Optional[Dict[str, Any]] = None,
123 rest_numeric_enums: bool = False,
124) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
125 """Transcodes a request into HTTP method, URI, body, and query parameters.
127 Args:
128 http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
129 request (Any): The protobuf or proto-plus request message.
130 required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
131 of required fields default values to merge into query parameters if missing.
132 rest_numeric_enums (bool): Whether to encode enums as integers.
134 Returns:
135 Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
136 - The raw transcoded request dictionary (containing keys like 'uri', 'method').
137 - The serialized request body JSON string, or None if no body.
138 - The query parameters dictionary.
139 """
140 if request is None:
141 raise TypeError("request cannot be None")
143 # Convert proto-plus message to its underlying protobuf message if needed
144 pb_request = getattr(request, "_pb", request)
146 transcoded_request = path_template.transcode(http_options, pb_request)
148 body_json = None
149 if transcoded_request.get("body") is not None:
150 body_json = json_format.MessageToJson(
151 transcoded_request["body"],
152 use_integers_for_enums=rest_numeric_enums,
153 )
155 query_params_json = {}
156 if transcoded_request.get("query_params") is not None:
157 query_params_json = json.loads(
158 json_format.MessageToJson(
159 transcoded_request["query_params"],
160 use_integers_for_enums=rest_numeric_enums,
161 )
162 )
164 # If required_fields_default_values is provided, we merge default values for missing
165 # required fields into the query parameters.
166 if required_fields_default_values:
167 for k, v in required_fields_default_values.items():
168 if k not in query_params_json:
169 query_params_json[k] = v
171 if rest_numeric_enums:
172 query_params_json["$alt"] = "json;enum-encoding=int"
174 return transcoded_request, body_json, query_params_json
177transcode = transcode_request