Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/utils.py: 22%

1""" 

2This module provides general utility functions used within Connexion. 

3""" 

4 

5import functools 

6import importlib 

7import inspect 

8import os 

9import sys 

10import typing 

11import typing as t 

12 

13import yaml 

14from starlette.routing import compile_path 

15 

16from connexion.exceptions import TypeValidationError 

17 

18if t.TYPE_CHECKING: 

19 from connexion.middleware.main import API 

20 

21 

22def boolean(s): 

23 """ 

24 Convert JSON/Swagger boolean value to Python, raise ValueError otherwise 

25 

26 >>> boolean('true') 

27 True 

28 

29 >>> boolean('false') 

30 False 

31 """ 

32 if isinstance(s, bool): 

33 return s 

34 elif not hasattr(s, "lower"): 

35 raise ValueError("Invalid boolean value") 

36 elif s.lower() == "true": 

37 return True 

38 elif s.lower() == "false": 

39 return False 

40 else: 

41 raise ValueError("Invalid boolean value") 

42 

43 

44# https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#data-types 

45TYPE_MAP: t.Dict[str, t.Any] = { 

46 "integer": int, 

47 "number": float, 

48 "string": str, 

49 "boolean": boolean, 

50 "array": list, 

51 "object": dict, 

52 "file": lambda x: x, # Don't cast files 

53} # map of swagger types to python types 

54 

55 

56def make_type(value: t.Any, type_: str, format_: t.Optional[str]) -> t.Any: 

57 """Cast a value to the type defined in the specification.""" 

58 # In OpenAPI, files are represented with string type and binary format 

59 if type_ == "string" and format_ == "binary": 

60 type_ = "file" 

61 

62 type_func = TYPE_MAP[type_] 

63 return type_func(value) 

64 

65 

66def deep_merge(a, b): 

67 """merges b into a 

68 in case of conflict the value from b is used 

69 """ 

70 for key in b: 

71 if key in a: 

72 if isinstance(a[key], dict) and isinstance(b[key], dict): 

73 deep_merge(a[key], b[key]) 

74 elif a[key] == b[key]: 

75 pass 

76 else: 

77 # b overwrites a 

78 a[key] = b[key] 

79 else: 

80 a[key] = b[key] 

81 return a 

82 

83 

84def deep_getattr(obj, attr): 

85 """ 

86 Recurses through an attribute chain to get the ultimate value. 

87 """ 

88 

89 attrs = attr.split(".") 

90 

91 return functools.reduce(getattr, attrs, obj) 

92 

93 

94def deep_get(obj, keys): 

95 """ 

96 Recurses through a nested object get a leaf value. 

97 

98 There are cases where the use of inheritance or polymorphism-- the use of allOf or 

99 oneOf keywords-- will cause the obj to be a list. In this case the keys will 

100 contain one or more strings containing integers. 

101 

102 :type obj: list or dict 

103 :type keys: list of strings 

104 """ 

105 if not keys: 

106 return obj 

107 

108 if isinstance(obj, list): 

109 return deep_get(obj[int(keys[0])], keys[1:]) 

110 else: 

111 return deep_get(obj[keys[0]], keys[1:]) 

112 

113 

114def get_function_from_name(function_name): 

115 """ 

116 Tries to get function by fully qualified name (e.g. "mymodule.myobj.myfunc") 

117 

118 :type function_name: str 

119 """ 

120 if function_name is None: 

121 raise ValueError("Empty function name") 

122 

123 if "." in function_name: 

124 module_name, attr_path = function_name.rsplit(".", 1) 

125 else: 

126 module_name = "" 

127 attr_path = function_name 

128 

129 module = None 

130 last_import_error = None 

131 

132 while not module: 

133 try: 

134 module = importlib.import_module(module_name) 

135 except ImportError as import_error: 

136 last_import_error = import_error 

137 if "." in module_name: 

138 module_name, attr_path1 = module_name.rsplit(".", 1) 

139 attr_path = f"{attr_path1}.{attr_path}" 

140 else: 

141 raise 

142 try: 

143 function = deep_getattr(module, attr_path) 

144 except AttributeError: 

145 if last_import_error: 

146 raise last_import_error 

147 else: 

148 raise 

149 return function 

150 

151 

152def is_json_mimetype(mimetype): 

153 """ 

154 :type mimetype: str 

155 :rtype: bool 

156 """ 

157 if mimetype is None: 

158 return False 

159 

160 maintype, subtype = mimetype.split("/") # type: str, str 

161 if ";" in subtype: 

162 subtype, parameter = subtype.split(";", maxsplit=1) 

163 return maintype == "application" and ( 

164 subtype == "json" or subtype.endswith("+json") 

165 ) 

166 

167 

168def all_json(mimetypes): 

169 """ 

170 Returns True if all mimetypes are serialized with json 

171 

172 :type mimetypes: list 

173 :rtype: bool 

174 

175 >>> all_json(['application/json']) 

176 True 

177 >>> all_json(['application/x.custom+json']) 

178 True 

179 >>> all_json([]) 

180 True 

181 >>> all_json(['application/xml']) 

182 False 

183 >>> all_json(['text/json']) 

184 False 

185 >>> all_json(['application/json', 'other/type']) 

186 False 

187 >>> all_json(['application/json', 'application/x.custom+json']) 

188 True 

189 """ 

190 return all(is_json_mimetype(mimetype) for mimetype in mimetypes) 

191 

192 

193def is_nullable(param_def): 

194 return param_def.get("schema", param_def).get("nullable", False) or param_def.get( 

195 "x-nullable", False 

196 ) # swagger2 

197 

198 

199def is_null(value): 

200 if hasattr(value, "strip") and value.strip() in ["null", "None"]: 

201 return True 

202 

203 if value is None: 

204 return True 

205 

206 return False 

207 

208 

209def has_coroutine(function, api=None): 

210 """ 

211 Checks if function is a coroutine. 

212 If ``function`` is a decorator (has a ``__wrapped__`` attribute) 

213 this function will also look at the wrapped function. 

214 """ 

215 

216 def iscorofunc(func): 

217 iscorofunc = inspect.iscoroutinefunction(func) 

218 while not iscorofunc and hasattr(func, "__wrapped__"): 

219 func = func.__wrapped__ 

220 iscorofunc = inspect.iscoroutinefunction(func) 

221 return iscorofunc 

222 

223 if api is None: 

224 return iscorofunc(function) 

225 

226 else: 

227 return any(iscorofunc(func) for func in (function, api.get_response)) 

228 

229 

230def yamldumper(openapi): 

231 """ 

232 Returns a nicely-formatted yaml spec. 

233 :param openapi: a spec dictionary. 

234 :return: a nicely-formatted, serialized yaml spec. 

235 """ 

236 

237 def should_use_block(value): 

238 char_list = ( 

239 "\u000a" # line feed 

240 "\u000d" # carriage return 

241 "\u001c" # file separator 

242 "\u001d" # group separator 

243 "\u001e" # record separator 

244 "\u0085" # next line 

245 "\u2028" # line separator 

246 "\u2029" # paragraph separator 

247 ) 

248 for c in char_list: 

249 if c in value: 

250 return True 

251 return False 

252 

253 def my_represent_scalar(self, tag, value, style=None): 

254 if should_use_block(value): 

255 style = "|" 

256 else: 

257 style = self.default_style 

258 

259 node = yaml.representer.ScalarNode(tag, value, style=style) 

260 if self.alias_key is not None: 

261 self.represented_objects[self.alias_key] = node 

262 return node 

263 

264 class NoAnchorDumper(yaml.dumper.SafeDumper): 

265 """A yaml Dumper that does not replace duplicate entries 

266 with yaml anchors. 

267 """ 

268 

269 def ignore_aliases(self, *args): 

270 return True 

271 

272 # Dump long lines as "|". 

273 yaml.representer.SafeRepresenter.represent_scalar = my_represent_scalar 

274 

275 return yaml.dump(openapi, allow_unicode=True, Dumper=NoAnchorDumper) 

276 

277 

278def not_installed_error(exc, *, msg=None): # pragma: no cover 

279 """Raises the ImportError when the module/object is actually called with a custom message.""" 

280 

281 def _delayed_error(*args, **kwargs): 

282 if msg is not None: 

283 raise type(exc)(msg).with_traceback(exc.__traceback__) 

284 raise exc 

285 

286 return _delayed_error 

287 

288 

289def extract_content_type( 

290 headers: t.Union[t.List[t.Tuple[bytes, bytes]], t.Dict[str, str]], 

291) -> t.Optional[str]: 

292 """Extract the mime type and encoding from the content type headers. 

293 

294 :param headers: Headers from ASGI scope 

295 

296 :return: The content type if available in headers, otherwise None 

297 """ 

298 content_type: t.Optional[str] = None 

299 

300 header_pairs_type = t.Collection[t.Tuple[t.Union[str, bytes], t.Union[str, bytes]]] 

301 header_pairs: header_pairs_type = headers.items() if isinstance(headers, dict) else headers # type: ignore 

302 for key, value in header_pairs: 

303 # Headers can always be decoded using latin-1: 

304 # https://stackoverflow.com/a/27357138/4098821 

305 if isinstance(key, bytes): 

306 decoded_key: str = key.decode("latin-1") 

307 else: 

308 decoded_key = key 

309 

310 if decoded_key.lower() == "content-type": 

311 if isinstance(value, bytes): 

312 content_type = value.decode("latin-1") 

313 else: 

314 content_type = value 

315 break 

316 

317 return content_type 

318 

319 

320def split_content_type( 

321 content_type: t.Optional[str], 

322) -> t.Tuple[t.Optional[str], t.Optional[str]]: 

323 """Split the content type in mime_type and encoding. Other parameters are ignored.""" 

324 mime_type, encoding = None, None 

325 

326 if content_type is None: 

327 return mime_type, encoding 

328 

329 # Check for parameters 

330 if ";" in content_type: 

331 mime_type, parameters = content_type.split(";", maxsplit=1) 

332 

333 # Find parameter describing the charset 

334 prefix = "charset=" 

335 for parameter in parameters.split(";"): 

336 if parameter.startswith(prefix): 

337 encoding = parameter[len(prefix) :] 

338 else: 

339 mime_type = content_type 

340 return mime_type, encoding 

341 

342 

343def coerce_type(param, value, parameter_type, parameter_name=None): 

344 # TODO: clean up 

345 TYPE_MAP = {"integer": int, "number": float, "boolean": boolean, "object": dict} 

346 

347 def make_type(value, type_literal): 

348 type_func = TYPE_MAP.get(type_literal) 

349 return type_func(value) 

350 

351 param_schema = param.get("schema", param) 

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

353 return None 

354 

355 param_type = param_schema.get("type") 

356 parameter_name = parameter_name if parameter_name else param.get("name") 

357 if param_type == "array": 

358 converted_params = [] 

359 if parameter_type == "header": 

360 value = value.split(",") 

361 for v in value: 

362 try: 

363 converted = make_type(v, param_schema["items"]["type"]) 

364 except (ValueError, TypeError): 

365 converted = v 

366 converted_params.append(converted) 

367 return converted_params 

368 elif param_type == "object": 

369 if param_schema.get("properties"): 

370 

371 def cast_leaves(d, schema): 

372 if type(d) is not dict: 

373 try: 

374 return make_type(d, schema["type"]) 

375 except (ValueError, TypeError): 

376 return d 

377 for k, v in d.items(): 

378 if k in schema["properties"]: 

379 d[k] = cast_leaves(v, schema["properties"][k]) 

380 return d 

381 

382 return cast_leaves(value, param_schema) 

383 return value 

384 else: 

385 try: 

386 return make_type(value, param_type) 

387 except ValueError: 

388 raise TypeValidationError(param_type, parameter_type, parameter_name) 

389 except TypeError: 

390 return value 

391 

392 

393@typing.no_type_check 

394def get_root_path(import_name: str) -> str: 

395 """Copied from Flask: 

396 https://github.com/pallets/flask/blob/836866dc19218832cf02f8b04911060ac92bfc0b/src/flask/helpers.py#L595 

397 

398 Find the root path of a package, or the path that contains a 

399 module. If it cannot be found, returns the current working 

400 directory. 

401 

402 :meta private: 

403 """ 

404 # Module already imported and has a file attribute. Use that first. 

405 mod = sys.modules.get(import_name) 

406 

407 if mod is not None and hasattr(mod, "__file__") and mod.__file__ is not None: 

408 return os.path.dirname(os.path.abspath(mod.__file__)) 

409 

410 # Next attempt: check the loader. 

411 try: 

412 spec = importlib.util.find_spec(import_name) 

413 

414 if spec is None: 

415 raise ValueError 

416 except (ImportError, ValueError): 

417 loader = None 

418 else: 

419 loader = spec.loader 

420 

421 # Loader does not exist or we're referring to an unloaded main 

422 # module or a main module without path (interactive sessions), go 

423 # with the current working directory. 

424 if loader is None: 

425 return os.getcwd() 

426 

427 if hasattr(loader, "get_filename"): 

428 filepath = loader.get_filename(import_name) # pyright: ignore 

429 else: 

430 # Fall back to imports. 

431 __import__(import_name) 

432 mod = sys.modules[import_name] 

433 filepath = getattr(mod, "__file__", None) 

434 

435 # If we don't have a file path it might be because it is a 

436 # namespace package. In this case pick the root path from the 

437 # first module that is contained in the package. 

438 if filepath is None: 

439 raise RuntimeError( 

440 "No root path can be found for the provided module" 

441 f" {import_name!r}. This can happen because the module" 

442 " came from an import hook that does not provide file" 

443 " name information or because it's a namespace package." 

444 " In this case the root path needs to be explicitly" 

445 " provided." 

446 ) 

447 

448 # filepath is import_name.py for a module, or __init__.py for a package. 

449 return os.path.dirname(os.path.abspath(filepath)) # type: ignore[no-any-return] 

450 

451 

452def inspect_function_arguments(function: t.Callable) -> t.Tuple[t.List[str], bool]: 

453 """ 

454 Returns the list of variables names of a function and if it 

455 accepts keyword arguments. 

456 """ 

457 parameters = inspect.signature(function).parameters 

458 bound_arguments = [ 

459 name 

460 for name, p in parameters.items() 

461 if p.kind not in (p.VAR_POSITIONAL, p.VAR_KEYWORD) 

462 ] 

463 has_kwargs = any(p.kind == p.VAR_KEYWORD for p in parameters.values()) 

464 return list(bound_arguments), has_kwargs 

465 

466 

467T = t.TypeVar("T") 

468 

469 

470@t.overload 

471def sort_routes(routes: t.List[str], *, key: None = None) -> t.List[str]: ... 

472 

473 

474@t.overload 

475def sort_routes(routes: t.List[T], *, key: t.Callable[[T], str]) -> t.List[T]: ... 

476 

477 

478def sort_routes(routes, *, key=None): 

479 """Sorts a list of routes from most specific to least specific. 

480 

481 See Starlette routing documentation and implementation as this function 

482 is aimed to sort according to that logic. 

483 - https://www.starlette.io/routing/#route-priority 

484 

485 The only difference is that a `path` component is appended to each route 

486 such that `/` is less specific than `/basepath` while they are technically 

487 not comparable. 

488 This is because it is also done by the `Mount` class internally: 

489 https://github.com/encode/starlette/blob/1c1043ca0ab7126419948b27f9d0a78270fd74e6/starlette/routing.py#L388 

490 

491 For example, from most to least specific: 

492 - /users/me 

493 - /users/{username}/projects/{project} 

494 - /users/{username} 

495 

496 :param routes: List of routes to sort 

497 :param key: Function to extract the path from a route if it is not a string 

498 

499 :return: List of routes sorted from most specific to least specific 

500 """ 

501 

502 class SortableRoute: 

503 def __init__(self, path: str) -> None: 

504 self.path = path.rstrip("/") 

505 if not self.path.endswith("/{path:path}"): 

506 self.path += "/{path:path}" 

507 self.path_regex, _, _ = compile_path(self.path) 

508 

509 def __lt__(self, other: "SortableRoute") -> bool: 

510 return bool(other.path_regex.match(self.path)) or self.path < other.path 

511 

512 return sorted(routes, key=lambda r: SortableRoute(key(r) if key else r)) 

513 

514 

515def sort_apis_by_basepath(apis: t.List["API"]) -> t.List["API"]: 

516 """Sorts a list of APIs by basepath. 

517 

518 :param apis: List of APIs to sort 

519 

520 :return: List of APIs sorted by basepath 

521 """ 

522 return sort_routes(apis, key=lambda api: api.base_path or "/") 

523 

524 

525def build_example_from_schema(schema): 

526 if "example" in schema: 

527 return schema["example"] 

528 

529 if "properties" in schema: 

530 # Recurse if schema is an object 

531 return { 

532 key: build_example_from_schema(value) 

533 for (key, value) in schema["properties"].items() 

534 } 

535 

536 if "items" in schema: 

537 # Recurse if schema is an array 

538 min_item_count = schema.get("minItems", 0) 

539 max_item_count = schema.get("maxItems") 

540 

541 if max_item_count is None or max_item_count >= min_item_count + 1: 

542 item_count = min_item_count + 1 

543 else: 

544 item_count = min_item_count 

545 

546 return [build_example_from_schema(schema["items"]) for n in range(item_count)] 

547 

548 try: 

549 from jsf import JSF 

550 except ImportError: 

551 return None 

552 

553 try: 

554 faker = JSF(schema) 

555 return faker.generate() 

556 except TypeError: 

557 

558 return None