Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/decorators/parameter.py: 23%

1""" 

2This module defines a decorator to convert request parameters to arguments for the view function. 

3""" 

4 

5import abc 

6import asyncio 

7import builtins 

8import functools 

9import keyword 

10import logging 

11import re 

12import typing as t 

13from copy import copy, deepcopy 

14 

15import inflection 

16 

17from connexion.context import context, operation 

18from connexion.frameworks.abstract import Framework 

19from connexion.http_facts import FORM_CONTENT_TYPES 

20from connexion.lifecycle import ConnexionRequest, WSGIRequest 

21from connexion.operations import AbstractOperation, Swagger2Operation 

22from connexion.utils import ( 

23 deep_merge, 

24 inspect_function_arguments, 

25 is_null, 

26 is_nullable, 

27 make_type, 

28) 

29 

30logger = logging.getLogger(__name__) 

31 

32CONTEXT_NAME = "context_" 

33 

34 

35class BaseParameterDecorator: 

36 def __init__( 

37 self, 

38 *, 

39 framework: t.Type[Framework], 

40 pythonic_params: bool = False, 

41 ) -> None: 

42 self.framework = framework 

43 self.sanitize_fn = pythonic if pythonic_params else sanitized 

44 

45 def _maybe_get_body( 

46 self, 

47 request: t.Union[WSGIRequest, ConnexionRequest], 

48 *, 

49 arguments: t.List[str], 

50 has_kwargs: bool, 

51 ) -> t.Any: 

52 body_name = self.sanitize_fn(operation.body_name(request.content_type)) 

53 # Pass form contents separately for Swagger2 for backward compatibility with 

54 # Connexion 2 Checking for body_name is not enough 

55 if (body_name in arguments or has_kwargs) or ( 

56 request.mimetype in FORM_CONTENT_TYPES 

57 and isinstance(operation, Swagger2Operation) 

58 ): 

59 return request.get_body() 

60 else: 

61 return None 

62 

63 @abc.abstractmethod 

64 def __call__(self, function: t.Callable) -> t.Callable: 

65 raise NotImplementedError 

66 

67 

68class SyncParameterDecorator(BaseParameterDecorator): 

69 def __call__(self, function: t.Callable) -> t.Callable: 

70 unwrapped_function = unwrap_decorators(function) 

71 arguments, has_kwargs = inspect_function_arguments(unwrapped_function) 

72 

73 @functools.wraps(function) 

74 def wrapper(request: WSGIRequest) -> t.Any: 

75 request_body = self._maybe_get_body( 

76 request, arguments=arguments, has_kwargs=has_kwargs 

77 ) 

78 

79 kwargs = prep_kwargs( 

80 request, 

81 request_body=request_body, 

82 files=request.files(), 

83 arguments=arguments, 

84 has_kwargs=has_kwargs, 

85 sanitize=self.sanitize_fn, 

86 ) 

87 

88 return function(**kwargs) 

89 

90 return wrapper 

91 

92 

93class AsyncParameterDecorator(BaseParameterDecorator): 

94 def __call__(self, function: t.Callable) -> t.Callable: 

95 unwrapped_function = unwrap_decorators(function) 

96 arguments, has_kwargs = inspect_function_arguments(unwrapped_function) 

97 

98 @functools.wraps(function) 

99 async def wrapper(request: ConnexionRequest) -> t.Any: 

100 request_body = self._maybe_get_body( 

101 request, arguments=arguments, has_kwargs=has_kwargs 

102 ) 

103 

104 while asyncio.iscoroutine(request_body): 

105 request_body = await request_body 

106 

107 kwargs = prep_kwargs( 

108 request, 

109 request_body=request_body, 

110 files=await request.files(), 

111 arguments=arguments, 

112 has_kwargs=has_kwargs, 

113 sanitize=self.sanitize_fn, 

114 ) 

115 

116 return await function(**kwargs) 

117 

118 return wrapper 

119 

120 

121def prep_kwargs( 

122 request: t.Union[WSGIRequest, ConnexionRequest], 

123 *, 

124 request_body: t.Any, 

125 files: t.Dict[str, t.Any], 

126 arguments: t.List[str], 

127 has_kwargs: bool, 

128 sanitize: t.Callable, 

129) -> dict: 

130 kwargs = get_arguments( 

131 operation, 

132 path_params=request.path_params, 

133 query_params=request.query_params, 

134 body=request_body, 

135 files=files, 

136 arguments=arguments, 

137 has_kwargs=has_kwargs, 

138 sanitize=sanitize, 

139 content_type=request.mimetype, 

140 ) 

141 

142 # optionally convert parameter variable names to un-shadowed, snake_case form 

143 kwargs = {sanitize(k): v for k, v in kwargs.items()} 

144 

145 # add context info (e.g. from security decorator) 

146 for key, value in context.items(): 

147 if has_kwargs or key in arguments: 

148 kwargs[key] = value 

149 else: 

150 logger.debug("Context parameter '%s' not in function arguments", key) 

151 # attempt to provide the request context to the function 

152 if CONTEXT_NAME in arguments: 

153 kwargs[CONTEXT_NAME] = context 

154 

155 return kwargs 

156 

157 

158def unwrap_decorators(function: t.Callable) -> t.Callable: 

159 """Unwrap decorators to return the original function.""" 

160 while hasattr(function, "__wrapped__"): 

161 function = function.__wrapped__ # type: ignore 

162 return function 

163 

164 

165def snake_and_shadow(name: str) -> str: 

166 """ 

167 Converts the given name into Pythonic form. Firstly it converts CamelCase names to snake_case. Secondly it looks to 

168 see if the name matches a known built-in and if it does it appends an underscore to the name. 

169 :param name: The parameter name 

170 """ 

171 snake = inflection.underscore(name) 

172 if snake in builtins.__dict__ or keyword.iskeyword(snake): 

173 return f"{snake}_" 

174 return snake 

175 

176 

177def sanitized(name: str) -> str: 

178 return name and re.sub( 

179 "^[^a-zA-Z_]+", "", re.sub("[^0-9a-zA-Z_]", "", re.sub(r"\[(?!])", "_", name)) 

180 ) 

181 

182 

183def pythonic(name: str) -> str: 

184 name = name and snake_and_shadow(name) 

185 return sanitized(name) 

186 

187 

188def get_arguments( 

189 operation: AbstractOperation, 

190 *, 

191 path_params: dict, 

192 query_params: dict, 

193 body: t.Any, 

194 files: dict, 

195 arguments: t.List[str], 

196 has_kwargs: bool, 

197 sanitize: t.Callable, 

198 content_type: str, 

199) -> t.Dict[str, t.Any]: 

200 """ 

201 get arguments for handler function 

202 """ 

203 ret = {} 

204 ret.update(_get_path_arguments(path_params, operation=operation, sanitize=sanitize)) 

205 ret.update( 

206 _get_query_arguments( 

207 query_params, 

208 operation=operation, 

209 arguments=arguments, 

210 has_kwargs=has_kwargs, 

211 sanitize=sanitize, 

212 ) 

213 ) 

214 

215 if operation.method.upper() == "TRACE": 

216 # TRACE requests MUST NOT include a body (RFC7231 section 4.3.8) 

217 return ret 

218 

219 ret.update( 

220 _get_body_argument( 

221 body, 

222 operation=operation, 

223 arguments=arguments, 

224 has_kwargs=has_kwargs, 

225 sanitize=sanitize, 

226 content_type=content_type, 

227 ) 

228 ) 

229 body_schema = operation.body_schema(content_type) 

230 ret.update(_get_file_arguments(files, arguments, body_schema, has_kwargs)) 

231 return ret 

232 

233 

234def _get_path_arguments( 

235 path_params: dict, *, operation: AbstractOperation, sanitize: t.Callable 

236) -> dict: 

237 """ 

238 Extract handler function arguments from path parameters 

239 """ 

240 kwargs = {} 

241 

242 path_definitions = { 

243 parameter["name"]: parameter 

244 for parameter in operation.parameters 

245 if parameter["in"] == "path" 

246 } 

247 

248 for name, value in path_params.items(): 

249 sanitized_key = sanitize(name) 

250 if name in path_definitions: 

251 kwargs[sanitized_key] = _get_val_from_param(value, path_definitions[name]) 

252 else: # Assume path params mechanism used for injection 

253 kwargs[sanitized_key] = value 

254 return kwargs 

255 

256 

257def _get_val_from_param(value: t.Any, param_definitions: t.Dict[str, dict]) -> t.Any: 

258 """Cast a value according to its definition in the specification.""" 

259 param_schema = param_definitions.get("schema", param_definitions) 

260 

261 if is_nullable(param_schema) and is_null(value): 

262 return None 

263 

264 if param_schema["type"] == "array": 

265 type_ = param_schema["items"]["type"] 

266 format_ = param_schema["items"].get("format") 

267 return [make_type(part, type_, format_) for part in value] 

268 else: 

269 type_ = param_schema["type"] 

270 format_ = param_schema.get("format") 

271 return make_type(value, type_, format_) 

272 

273 

274def _get_query_arguments( 

275 query_params: dict, 

276 *, 

277 operation: AbstractOperation, 

278 arguments: t.List[str], 

279 has_kwargs: bool, 

280 sanitize: t.Callable, 

281) -> dict: 

282 """ 

283 extract handler function arguments from the query parameters 

284 """ 

285 query_definitions = { 

286 parameter["name"]: parameter 

287 for parameter in operation.parameters 

288 if parameter["in"] == "query" 

289 } 

290 

291 default_query_params = _get_query_defaults(query_definitions) 

292 

293 query_arguments = deepcopy(default_query_params) 

294 query_arguments = deep_merge(query_arguments, query_params) 

295 return _query_args_helper( 

296 query_definitions, query_arguments, arguments, has_kwargs, sanitize 

297 ) 

298 

299 

300def _get_query_defaults(query_definitions: t.Dict[str, dict]) -> t.Dict[str, t.Any]: 

301 """Get the default values for the query parameter from the parameter definition.""" 

302 defaults = {} 

303 for k, v in query_definitions.items(): 

304 try: 

305 if "default" in v: 

306 defaults[k] = v["default"] 

307 elif v["schema"]["type"] == "object": 

308 defaults[k] = _get_default_obj(v["schema"]) 

309 else: 

310 defaults[k] = v["schema"]["default"] 

311 except KeyError: 

312 pass 

313 return defaults 

314 

315 

316def _get_default_obj(schema: dict) -> dict: 

317 try: 

318 return deepcopy(schema["default"]) 

319 except KeyError: 

320 properties = schema.get("properties", {}) 

321 return _build_default_obj_recursive(properties, {}) 

322 

323 

324def _build_default_obj_recursive(properties: dict, default_object: dict) -> dict: 

325 """takes disparate and nested default keys, and builds up a default object""" 

326 for name, property_ in properties.items(): 

327 if "default" in property_ and name not in default_object: 

328 default_object[name] = copy(property_["default"]) 

329 elif property_.get("type") == "object" and "properties" in property_: 

330 default_object.setdefault(name, {}) 

331 default_object[name] = _build_default_obj_recursive( 

332 property_["properties"], default_object[name] 

333 ) 

334 return default_object 

335 

336 

337def _query_args_helper( 

338 query_definitions: dict, 

339 query_arguments: dict, 

340 function_arguments: t.List[str], 

341 has_kwargs: bool, 

342 sanitize: t.Callable, 

343) -> dict: 

344 result = {} 

345 for key, value in query_arguments.items(): 

346 sanitized_key = sanitize(key) 

347 if not has_kwargs and sanitized_key not in function_arguments: 

348 logger.debug( 

349 "Query Parameter '%s' (sanitized: '%s') not in function arguments", 

350 key, 

351 sanitized_key, 

352 ) 

353 else: 

354 logger.debug( 

355 "Query Parameter '%s' (sanitized: '%s') in function arguments", 

356 key, 

357 sanitized_key, 

358 ) 

359 try: 

360 query_defn = query_definitions[key] 

361 except KeyError: # pragma: no cover 

362 logger.error( 

363 "Function argument '%s' (non-sanitized: %s) not defined in specification", 

364 sanitized_key, 

365 key, 

366 ) 

367 else: 

368 logger.debug("%s is a %s", key, query_defn) 

369 result.update({sanitized_key: _get_val_from_param(value, query_defn)}) 

370 return result 

371 

372 

373def _get_body_argument( 

374 body: t.Any, 

375 *, 

376 operation: AbstractOperation, 

377 arguments: t.List[str], 

378 has_kwargs: bool, 

379 sanitize: t.Callable, 

380 content_type: str, 

381) -> dict: 

382 if len(arguments) <= 0 and not has_kwargs: 

383 return {} 

384 

385 if not operation.is_request_body_defined: 

386 return {} 

387 

388 body_name = sanitize(operation.body_name(content_type)) 

389 

390 if content_type in FORM_CONTENT_TYPES: 

391 result = _get_body_argument_form( 

392 body, operation=operation, content_type=content_type 

393 ) 

394 

395 # Unpack form values for Swagger for compatibility with Connexion 2 behavior 

396 if content_type in FORM_CONTENT_TYPES and isinstance( 

397 operation, Swagger2Operation 

398 ): 

399 if has_kwargs: 

400 return result 

401 else: 

402 return { 

403 sanitize(name): value 

404 for name, value in result.items() 

405 if sanitize(name) in arguments 

406 } 

407 else: 

408 result = _get_body_argument_json( 

409 body, operation=operation, content_type=content_type 

410 ) 

411 

412 if body_name in arguments or has_kwargs: 

413 return {body_name: result} 

414 

415 return {} 

416 

417 

418def _get_body_argument_json( 

419 body: t.Any, *, operation: AbstractOperation, content_type: str 

420) -> t.Any: 

421 # if the body came in null, and the schema says it can be null, we decide 

422 # to include no value for the body argument, rather than the default body 

423 if is_nullable(operation.body_schema(content_type)) and is_null(body): 

424 return None 

425 

426 if body is None: 

427 default_body = operation.body_schema(content_type).get("default", {}) 

428 return deepcopy(default_body) 

429 

430 return body 

431 

432 

433def _get_body_argument_form( 

434 body: dict, *, operation: AbstractOperation, content_type: str 

435) -> dict: 

436 # now determine the actual value for the body (whether it came in or is default) 

437 default_body = operation.body_schema(content_type).get("default", {}) 

438 body_props = { 

439 k: {"schema": v} 

440 for k, v in operation.body_schema(content_type).get("properties", {}).items() 

441 } 

442 

443 # by OpenAPI specification `additionalProperties` defaults to `true` 

444 # see: https://github.com/OAI/OpenAPI-Specification/blame/3.0.2/versions/3.0.2.md#L2305 

445 additional_props = operation.body_schema().get("additionalProperties", True) 

446 

447 body_arg = deepcopy(default_body) 

448 body_arg.update(body or {}) 

449 

450 if body_props or additional_props: 

451 return _get_typed_body_values(body_arg, body_props, additional_props) 

452 

453 return {} 

454 

455 

456def _get_typed_body_values(body_arg, body_props, additional_props): 

457 """ 

458 Return a copy of the provided body_arg dictionary 

459 whose values will have the appropriate types 

460 as defined in the provided schemas. 

461 

462 :type body_arg: type dict 

463 :type body_props: dict 

464 :type additional_props: dict|bool 

465 :rtype: dict 

466 """ 

467 additional_props_defn = ( 

468 {"schema": additional_props} if isinstance(additional_props, dict) else None 

469 ) 

470 res = {} 

471 

472 for key, value in body_arg.items(): 

473 try: 

474 prop_defn = body_props[key] 

475 res[key] = _get_val_from_param(value, prop_defn) 

476 except KeyError: # pragma: no cover 

477 if not additional_props: 

478 logger.error(f"Body property '{key}' not defined in body schema") 

479 continue 

480 if additional_props_defn is not None: 

481 value = _get_val_from_param(value, additional_props_defn) 

482 res[key] = value 

483 

484 return res 

485 

486 

487def _get_file_arguments(files, arguments, body_schema: dict, has_kwargs=False): 

488 results = {} 

489 for k, v in files.items(): 

490 if not (k in arguments or has_kwargs): 

491 continue 

492 if body_schema.get("properties", {}).get(k, {}).get("type") != "array": 

493 v = v[0] 

494 results[k] = v 

495 

496 return results