Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/uri_parsing.py: 67%
165 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1"""
2This module defines URIParsers which parse query and path parameters according to OpenAPI
3serialization rules.
4"""
6import abc
7import json
8import logging
9import re
11from connexion.exceptions import TypeValidationError
12from connexion.utils import all_json, coerce_type, deep_merge
14logger = logging.getLogger("connexion.decorators.uri_parsing")
16QUERY_STRING_DELIMITERS = {
17 "spaceDelimited": " ",
18 "pipeDelimited": "|",
19 "simple": ",",
20 "form": ",",
21}
24class AbstractURIParser(metaclass=abc.ABCMeta):
25 parsable_parameters = ["query", "path"]
27 def __init__(self, param_defns, body_defn):
28 """
29 a URI parser is initialized with parameter definitions.
30 When called with a request object, it handles array types in the URI
31 both in the path and query according to the spec.
32 Some examples include:
33 - https://mysite.fake/in/path/1,2,3/ # path parameters
34 - https://mysite.fake/?in_query=a,b,c # simple query params
35 - https://mysite.fake/?in_query=a|b|c # various separators
36 - https://mysite.fake/?in_query=a&in_query=b,c # complex query params
37 """
38 self._param_defns = {
39 p["name"]: p for p in param_defns if p["in"] in self.parsable_parameters
40 }
41 self._body_schema = body_defn.get("schema", {})
42 self._body_encoding = body_defn.get("encoding", {})
44 @property
45 @abc.abstractmethod
46 def param_defns(self):
47 """
48 returns the parameter definitions by name
49 """
51 @property
52 @abc.abstractmethod
53 def param_schemas(self):
54 """
55 returns the parameter schemas by name
56 """
58 def __repr__(self):
59 """
60 :rtype: str
61 """
62 return "<{classname}>".format(
63 classname=self.__class__.__name__
64 ) # pragma: no cover
66 @abc.abstractmethod
67 def resolve_form(self, form_data):
68 """Resolve cases where form parameters are provided multiple times."""
70 @abc.abstractmethod
71 def resolve_query(self, query_data):
72 """Resolve cases where query parameters are provided multiple times."""
74 @abc.abstractmethod
75 def resolve_path(self, path):
76 """Resolve cases where path parameters include lists"""
78 @abc.abstractmethod
79 def _resolve_param_duplicates(self, values, param_defn, _in):
80 """Resolve cases where query parameters are provided multiple times.
81 For example, if the query string is '?a=1,2,3&a=4,5,6' the value of
82 `a` could be "4,5,6", or "1,2,3" or "1,2,3,4,5,6" depending on the
83 implementation.
84 """
86 @abc.abstractmethod
87 def _split(self, value, param_defn, _in):
88 """
89 takes a string, a parameter definition, and a parameter type
90 and returns an array that has been constructed according to
91 the parameter definition.
92 """
94 def resolve_params(self, params, _in):
95 """
96 takes a dict of parameters, and resolves the values into
97 the correct array type handling duplicate values, and splitting
98 based on the collectionFormat defined in the spec.
99 """
100 resolved_param = {}
101 for k, values in params.items():
102 param_defn = self.param_defns.get(k)
103 param_schema = self.param_schemas.get(k)
105 if not (param_defn or param_schema):
106 # rely on validation
107 resolved_param[k] = values
108 continue
110 if _in == "path":
111 # multiple values in a path is impossible
112 values = [values]
114 if param_schema and param_schema["type"] == "array":
115 # resolve variable re-assignment, handle explode
116 values = self._resolve_param_duplicates(values, param_defn, _in)
117 # handle array styles
118 resolved_param[k] = self._split(values, param_defn, _in)
119 else:
120 resolved_param[k] = values[-1]
122 try:
123 resolved_param[k] = coerce_type(
124 param_defn, resolved_param[k], "parameter", k
125 )
126 except TypeValidationError:
127 pass
129 return resolved_param
132class OpenAPIURIParser(AbstractURIParser):
133 style_defaults = {
134 "path": "simple",
135 "header": "simple",
136 "query": "form",
137 "cookie": "form",
138 "form": "form",
139 }
141 @property
142 def param_defns(self):
143 return self._param_defns
145 @property
146 def form_defns(self):
147 return {k: v for k, v in self._body_schema.get("properties", {}).items()}
149 @property
150 def param_schemas(self):
151 return {k: v.get("schema", {}) for k, v in self.param_defns.items()}
153 def resolve_form(self, form_data):
154 if self._body_schema is None or self._body_schema.get("type") != "object":
155 return form_data
156 for k in form_data:
157 encoding = self._body_encoding.get(k, {"style": "form"})
158 defn = self.form_defns.get(k, {})
159 # TODO support more form encoding styles
160 form_data[k] = self._resolve_param_duplicates(
161 form_data[k], encoding, "form"
162 )
163 if defn and defn["type"] == "array":
164 form_data[k] = self._split(form_data[k], encoding, "form")
165 elif "contentType" in encoding and all_json([encoding.get("contentType")]):
166 form_data[k] = json.loads(form_data[k])
167 form_data[k] = coerce_type(defn, form_data[k], "requestBody", k)
168 return form_data
170 def _make_deep_object(self, k, v):
171 """consumes keys, value pairs like (a[foo][bar], "baz")
172 returns (a, {"foo": {"bar": "baz"}}}, is_deep_object)
173 """
174 root_key = None
175 if k in self.param_schemas.keys():
176 return k, v, False
177 else:
178 for key in self.param_schemas.keys():
179 if k.startswith(key) and "[" in k:
180 root_key = key.replace(k, "")
182 if not root_key:
183 root_key = k.split("[", 1)[0]
184 if k == root_key:
185 return k, v, False
187 if not self._is_deep_object_style_param(root_key):
188 return k, v, False
190 key_path = re.findall(r"\[([^\[\]]*)\]", k)
191 root = prev = node = {}
192 for k in key_path:
193 node[k] = {}
194 prev = node
195 node = node[k]
196 prev[k] = v[0]
197 return root_key, [root], True
199 def _is_deep_object_style_param(self, param_name):
200 default_style = self.style_defaults["query"]
201 style = self.param_defns.get(param_name, {}).get("style", default_style)
202 return style == "deepObject"
204 def _preprocess_deep_objects(self, query_data):
205 """deep objects provide a way of rendering nested objects using query
206 parameters.
207 """
208 deep = [self._make_deep_object(k, v) for k, v in query_data.items()]
209 root_keys = [k for k, v, is_deep_object in deep]
210 ret = dict.fromkeys(root_keys, [{}])
211 for k, v, is_deep_object in deep:
212 if is_deep_object:
213 ret[k] = [deep_merge(v[0], ret[k][0])]
214 else:
215 ret[k] = v
216 return ret
218 def resolve_query(self, query_data):
219 query_data = self._preprocess_deep_objects(query_data)
220 return self.resolve_params(query_data, "query")
222 def resolve_path(self, path_data):
223 return self.resolve_params(path_data, "path")
225 @staticmethod
226 def _resolve_param_duplicates(values, param_defn, _in):
227 """Resolve cases where query parameters are provided multiple times.
228 The default behavior is to use the first-defined value.
229 For example, if the query string is '?a=1,2,3&a=4,5,6' the value of
230 `a` would be "4,5,6".
231 However, if 'explode' is 'True' then the duplicate values
232 are concatenated together and `a` would be "1,2,3,4,5,6".
233 """
234 default_style = OpenAPIURIParser.style_defaults[_in]
235 style = param_defn.get("style", default_style)
236 delimiter = QUERY_STRING_DELIMITERS.get(style, ",")
237 is_form = style == "form"
238 explode = param_defn.get("explode", is_form)
239 if explode:
240 return delimiter.join(values)
242 # default to last defined value
243 return values[-1]
245 @staticmethod
246 def _split(value, param_defn, _in):
247 default_style = OpenAPIURIParser.style_defaults[_in]
248 style = param_defn.get("style", default_style)
249 delimiter = QUERY_STRING_DELIMITERS.get(style, ",")
250 return value.split(delimiter)
253class Swagger2URIParser(AbstractURIParser):
254 """
255 Adheres to the Swagger2 spec,
256 Assumes that the last defined query parameter should be used.
257 """
259 parsable_parameters = ["query", "path", "formData"]
261 @property
262 def param_defns(self):
263 return self._param_defns
265 @property
266 def param_schemas(self):
267 return self._param_defns # swagger2 conflates defn and schema
269 def resolve_form(self, form_data):
270 return self.resolve_params(form_data, "form")
272 def resolve_query(self, query_data):
273 return self.resolve_params(query_data, "query")
275 def resolve_path(self, path_data):
276 return self.resolve_params(path_data, "path")
278 @staticmethod
279 def _resolve_param_duplicates(values, param_defn, _in):
280 """Resolve cases where query parameters are provided multiple times.
281 The default behavior is to use the first-defined value.
282 For example, if the query string is '?a=1,2,3&a=4,5,6' the value of
283 `a` would be "4,5,6".
284 However, if 'collectionFormat' is 'multi' then the duplicate values
285 are concatenated together and `a` would be "1,2,3,4,5,6".
286 """
287 if param_defn.get("collectionFormat") == "multi":
288 return ",".join(values)
289 # default to last defined value
290 return values[-1]
292 @staticmethod
293 def _split(value, param_defn, _in):
294 if param_defn.get("collectionFormat") == "pipes":
295 return value.split("|")
296 return value.split(",")
299class FirstValueURIParser(Swagger2URIParser):
300 """
301 Adheres to the Swagger2 spec
302 Assumes that the first defined query parameter should be used
303 """
305 @staticmethod
306 def _resolve_param_duplicates(values, param_defn, _in):
307 """Resolve cases where query parameters are provided multiple times.
308 The default behavior is to use the first-defined value.
309 For example, if the query string is '?a=1,2,3&a=4,5,6' the value of
310 `a` would be "1,2,3".
311 However, if 'collectionFormat' is 'multi' then the duplicate values
312 are concatenated together and `a` would be "1,2,3,4,5,6".
313 """
314 if param_defn.get("collectionFormat") == "multi":
315 return ",".join(values)
316 # default to first defined value
317 return values[0]
320class AlwaysMultiURIParser(Swagger2URIParser):
321 """
322 Does not adhere to the Swagger2 spec, but is backwards compatible with
323 connexion behavior in version 1.4.2
324 """
326 @staticmethod
327 def _resolve_param_duplicates(values, param_defn, _in):
328 """Resolve cases where query parameters are provided multiple times.
329 The default behavior is to join all provided parameters together.
330 For example, if the query string is '?a=1,2,3&a=4,5,6' the value of
331 `a` would be "1,2,3,4,5,6".
332 """
333 if param_defn.get("collectionFormat") == "pipes":
334 return "|".join(values)
335 return ",".join(values)