Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/uri_parsing.py: 67%

1""" 

2This module defines URIParsers which parse query and path parameters according to OpenAPI 

3serialization rules. 

4""" 

5 

6import abc 

7import json 

8import logging 

9import re 

10 

11from connexion.exceptions import TypeValidationError 

12from connexion.utils import all_json, coerce_type, deep_merge 

13 

14logger = logging.getLogger("connexion.decorators.uri_parsing") 

15 

16QUERY_STRING_DELIMITERS = { 

17 "spaceDelimited": " ", 

18 "pipeDelimited": "|", 

19 "simple": ",", 

20 "form": ",", 

21} 

22 

23 

24class AbstractURIParser(metaclass=abc.ABCMeta): 

25 parsable_parameters = ["query", "path"] 

26 

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", {}) 

43 

44 @property 

45 @abc.abstractmethod 

46 def param_defns(self): 

47 """ 

48 returns the parameter definitions by name 

49 """ 

50 

51 @property 

52 @abc.abstractmethod 

53 def param_schemas(self): 

54 """ 

55 returns the parameter schemas by name 

56 """ 

57 

58 def __repr__(self): 

59 """ 

60 :rtype: str 

61 """ 

62 return "<{classname}>".format( 

63 classname=self.__class__.__name__ 

64 ) # pragma: no cover 

65 

66 @abc.abstractmethod 

67 def resolve_form(self, form_data): 

68 """Resolve cases where form parameters are provided multiple times.""" 

69 

70 @abc.abstractmethod 

71 def resolve_query(self, query_data): 

72 """Resolve cases where query parameters are provided multiple times.""" 

73 

74 @abc.abstractmethod 

75 def resolve_path(self, path): 

76 """Resolve cases where path parameters include lists""" 

77 

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 """ 

85 

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 """ 

93 

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) 

104 

105 if not (param_defn or param_schema): 

106 # rely on validation 

107 resolved_param[k] = values 

108 continue 

109 

110 if _in == "path": 

111 # multiple values in a path is impossible 

112 values = [values] 

113 

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] 

121 

122 try: 

123 resolved_param[k] = coerce_type( 

124 param_defn, resolved_param[k], "parameter", k 

125 ) 

126 except TypeValidationError: 

127 pass 

128 

129 return resolved_param 

130 

131 

132class OpenAPIURIParser(AbstractURIParser): 

133 style_defaults = { 

134 "path": "simple", 

135 "header": "simple", 

136 "query": "form", 

137 "cookie": "form", 

138 "form": "form", 

139 } 

140 

141 @property 

142 def param_defns(self): 

143 return self._param_defns 

144 

145 @property 

146 def form_defns(self): 

147 return {k: v for k, v in self._body_schema.get("properties", {}).items()} 

148 

149 @property 

150 def param_schemas(self): 

151 return {k: v.get("schema", {}) for k, v in self.param_defns.items()} 

152 

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 "contentType" in encoding and all_json([encoding.get("contentType")]): 

164 form_data[k] = json.loads(form_data[k]) 

165 elif defn and defn["type"] == "array": 

166 form_data[k] = self._split(form_data[k], encoding, "form") 

167 form_data[k] = coerce_type(defn, form_data[k], "requestBody", k) 

168 return form_data 

169 

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, "") 

181 

182 if not root_key: 

183 root_key = k.split("[", 1)[0] 

184 if k == root_key: 

185 return k, v, False 

186 

187 if not self._is_deep_object_style_param(root_key): 

188 return k, v, False 

189 

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 

198 

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" 

203 

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 

217 

218 def resolve_query(self, query_data): 

219 query_data = self._preprocess_deep_objects(query_data) 

220 return self.resolve_params(query_data, "query") 

221 

222 def resolve_path(self, path_data): 

223 return self.resolve_params(path_data, "path") 

224 

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) 

241 

242 # default to last defined value 

243 return values[-1] 

244 

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) 

251 

252 

253class Swagger2URIParser(AbstractURIParser): 

254 """ 

255 Adheres to the Swagger2 spec, 

256 Assumes that the last defined query parameter should be used. 

257 """ 

258 

259 parsable_parameters = ["query", "path", "formData"] 

260 

261 @property 

262 def param_defns(self): 

263 return self._param_defns 

264 

265 @property 

266 def param_schemas(self): 

267 return self._param_defns # swagger2 conflates defn and schema 

268 

269 def resolve_form(self, form_data): 

270 return self.resolve_params(form_data, "form") 

271 

272 def resolve_query(self, query_data): 

273 return self.resolve_params(query_data, "query") 

274 

275 def resolve_path(self, path_data): 

276 return self.resolve_params(path_data, "path") 

277 

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] 

291 

292 @staticmethod 

293 def _split(value, param_defn, _in): 

294 if param_defn.get("collectionFormat") == "pipes": 

295 return value.split("|") 

296 return value.split(",") 

297 

298 

299class FirstValueURIParser(Swagger2URIParser): 

300 """ 

301 Adheres to the Swagger2 spec 

302 Assumes that the first defined query parameter should be used 

303 """ 

304 

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] 

318 

319 

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 """ 

325 

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)