Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/flask_restx/api.py: 21%

1import difflib 

2import inspect 

3from itertools import chain 

4import logging 

5import operator 

6import re 

7import sys 

8import warnings 

9 

10from collections import OrderedDict 

11from functools import wraps, partial 

12from types import MethodType 

13 

14from flask import url_for, request, current_app 

15from flask import make_response as original_flask_make_response 

16 

17from flask.signals import got_request_exception 

18 

19from jsonschema import RefResolver 

20 

21from werkzeug.utils import cached_property 

22from werkzeug.datastructures import Headers 

23from werkzeug.exceptions import ( 

24 HTTPException, 

25 MethodNotAllowed, 

26 NotFound, 

27 NotAcceptable, 

28 InternalServerError, 

29) 

30 

31from . import apidoc 

32from .mask import ParseError, MaskError 

33from .namespace import Namespace 

34from .postman import PostmanCollectionV1 

35from .resource import Resource 

36from .swagger import Swagger 

37from .utils import ( 

38 default_id, 

39 camel_to_dash, 

40 unpack, 

41 import_check_view_func, 

42 BaseResponse, 

43) 

44from .representations import output_json 

45from ._http import HTTPStatus 

46 

47endpoint_from_view_func = import_check_view_func() 

48 

49 

50RE_RULES = re.compile("(<.*>)") 

51 

52# List headers that should never be handled by Flask-RESTX 

53HEADERS_BLACKLIST = ("Content-Length",) 

54 

55DEFAULT_REPRESENTATIONS = [("application/json", output_json)] 

56 

57log = logging.getLogger(__name__) 

58 

59 

60class Api(object): 

61 """ 

62 The main entry point for the application. 

63 You need to initialize it with a Flask Application: :: 

64 

65 >>> app = Flask(__name__) 

66 >>> api = Api(app) 

67 

68 Alternatively, you can use :meth:`init_app` to set the Flask application 

69 after it has been constructed. 

70 

71 The endpoint parameter prefix all views and resources: 

72 

73 - The API root/documentation will be ``{endpoint}.root`` 

74 - A resource registered as 'resource' will be available as ``{endpoint}.resource`` 

75 

76 :param flask.Flask|flask.Blueprint app: the Flask application object or a Blueprint 

77 :param str version: The API version (used in Swagger documentation) 

78 :param str title: The API title (used in Swagger documentation) 

79 :param str description: The API description (used in Swagger documentation) 

80 :param str terms_url: The API terms page URL (used in Swagger documentation) 

81 :param str contact: A contact email for the API (used in Swagger documentation) 

82 :param str license: The license associated to the API (used in Swagger documentation) 

83 :param str license_url: The license page URL (used in Swagger documentation) 

84 :param str endpoint: The API base endpoint (default to 'api). 

85 :param str default: The default namespace base name (default to 'default') 

86 :param str default_label: The default namespace label (used in Swagger documentation) 

87 :param str default_mediatype: The default media type to return 

88 :param bool validate: Whether or not the API should perform input payload validation. 

89 :param bool ordered: Whether or not preserve order models and marshalling. 

90 :param str doc: The documentation path. If set to a false value, documentation is disabled. 

91 (Default to '/') 

92 :param list decorators: Decorators to attach to every resource 

93 :param bool catch_all_404s: Use :meth:`handle_error` 

94 to handle 404 errors throughout your app 

95 :param dict authorizations: A Swagger Authorizations declaration as dictionary 

96 :param bool serve_challenge_on_401: Serve basic authentication challenge with 401 

97 responses (default 'False') 

98 :param FormatChecker format_checker: A jsonschema.FormatChecker object that is hooked into 

99 the Model validator. A default or a custom FormatChecker can be provided (e.g., with custom 

100 checkers), otherwise the default action is to not enforce any format validation. 

101 :param url_scheme: If set to a string (e.g. http, https), then the specs_url and base_url will explicitly use this 

102 scheme regardless of how the application is deployed. This is necessary for some deployments behind a reverse 

103 proxy. 

104 :param str default_swagger_filename: The default swagger filename. 

105 """ 

106 

107 def __init__( 

108 self, 

109 app=None, 

110 version="1.0", 

111 title=None, 

112 description=None, 

113 terms_url=None, 

114 license=None, 

115 license_url=None, 

116 contact=None, 

117 contact_url=None, 

118 contact_email=None, 

119 authorizations=None, 

120 security=None, 

121 doc="/", 

122 default_id=default_id, 

123 default="default", 

124 default_label="Default namespace", 

125 validate=None, 

126 tags=None, 

127 prefix="", 

128 ordered=False, 

129 default_mediatype="application/json", 

130 decorators=None, 

131 catch_all_404s=False, 

132 serve_challenge_on_401=False, 

133 format_checker=None, 

134 url_scheme=None, 

135 default_swagger_filename="swagger.json", 

136 **kwargs 

137 ): 

138 self.version = version 

139 self.title = title or "API" 

140 self.description = description 

141 self.terms_url = terms_url 

142 self.contact = contact 

143 self.contact_email = contact_email 

144 self.contact_url = contact_url 

145 self.license = license 

146 self.license_url = license_url 

147 self.authorizations = authorizations 

148 self.security = security 

149 self.default_id = default_id 

150 self.ordered = ordered 

151 self._validate = validate 

152 self._doc = doc 

153 self._doc_view = None 

154 self._default_error_handler = None 

155 self.tags = tags or [] 

156 

157 self.error_handlers = OrderedDict( 

158 { 

159 ParseError: mask_parse_error_handler, 

160 MaskError: mask_error_handler, 

161 } 

162 ) 

163 self._schema = None 

164 self.models = {} 

165 self._refresolver = None 

166 self.format_checker = format_checker 

167 self.namespaces = [] 

168 self.default_swagger_filename = default_swagger_filename 

169 

170 self.ns_paths = dict() 

171 

172 self.representations = OrderedDict(DEFAULT_REPRESENTATIONS) 

173 self.urls = {} 

174 self.prefix = prefix 

175 self.default_mediatype = default_mediatype 

176 self.decorators = decorators if decorators else [] 

177 self.catch_all_404s = catch_all_404s 

178 self.serve_challenge_on_401 = serve_challenge_on_401 

179 self.blueprint_setup = None 

180 self.endpoints = set() 

181 self.resources = [] 

182 self.app = None 

183 self.blueprint = None 

184 # must come after self.app initialisation to prevent __getattr__ recursion 

185 # in self._configure_namespace_logger 

186 self.default_namespace = self.namespace( 

187 default, 

188 default_label, 

189 endpoint="{0}-declaration".format(default), 

190 validate=validate, 

191 api=self, 

192 path="/", 

193 ) 

194 self.url_scheme = url_scheme 

195 if app is not None: 

196 self.app = app 

197 self.init_app(app) 

198 # super(Api, self).__init__(app, **kwargs) 

199 

200 def init_app(self, app, **kwargs): 

201 """ 

202 Allow to lazy register the API on a Flask application:: 

203 

204 >>> app = Flask(__name__) 

205 >>> api = Api() 

206 >>> api.init_app(app) 

207 

208 :param flask.Flask app: the Flask application object 

209 :param str title: The API title (used in Swagger documentation) 

210 :param str description: The API description (used in Swagger documentation) 

211 :param str terms_url: The API terms page URL (used in Swagger documentation) 

212 :param str contact: A contact email for the API (used in Swagger documentation) 

213 :param str license: The license associated to the API (used in Swagger documentation) 

214 :param str license_url: The license page URL (used in Swagger documentation) 

215 :param url_scheme: If set to a string (e.g. http, https), then the specs_url and base_url will explicitly use 

216 this scheme regardless of how the application is deployed. This is necessary for some deployments behind a 

217 reverse proxy. 

218 """ 

219 self.app = app 

220 self.title = kwargs.get("title", self.title) 

221 self.description = kwargs.get("description", self.description) 

222 self.terms_url = kwargs.get("terms_url", self.terms_url) 

223 self.contact = kwargs.get("contact", self.contact) 

224 self.contact_url = kwargs.get("contact_url", self.contact_url) 

225 self.contact_email = kwargs.get("contact_email", self.contact_email) 

226 self.license = kwargs.get("license", self.license) 

227 self.license_url = kwargs.get("license_url", self.license_url) 

228 self.url_scheme = kwargs.get("url_scheme", self.url_scheme) 

229 self._add_specs = kwargs.get("add_specs", True) 

230 self._register_specs(app) 

231 self._register_doc(app) 

232 

233 # If app is a blueprint, defer the initialization 

234 try: 

235 app.record(self._deferred_blueprint_init) 

236 # Flask.Blueprint has a 'record' attribute, Flask.Api does not 

237 except AttributeError: 

238 self._init_app(app) 

239 else: 

240 self.blueprint = app 

241 

242 def _init_app(self, app): 

243 """ 

244 Perform initialization actions with the given :class:`flask.Flask` object. 

245 

246 :param flask.Flask app: The flask application object 

247 """ 

248 app.handle_exception = partial(self.error_router, app.handle_exception) 

249 app.handle_user_exception = partial( 

250 self.error_router, app.handle_user_exception 

251 ) 

252 

253 if len(self.resources) > 0: 

254 for resource, namespace, urls, kwargs in self.resources: 

255 self._register_view(app, resource, namespace, *urls, **kwargs) 

256 

257 for ns in self.namespaces: 

258 self._configure_namespace_logger(app, ns) 

259 

260 self._register_apidoc(app) 

261 self._validate = ( 

262 self._validate 

263 if self._validate is not None 

264 else app.config.get("RESTX_VALIDATE", False) 

265 ) 

266 app.config.setdefault("RESTX_MASK_HEADER", "X-Fields") 

267 app.config.setdefault("RESTX_MASK_SWAGGER", True) 

268 app.config.setdefault("RESTX_INCLUDE_ALL_MODELS", False) 

269 

270 # check for deprecated config variable names 

271 if "ERROR_404_HELP" in app.config: 

272 app.config["RESTX_ERROR_404_HELP"] = app.config["ERROR_404_HELP"] 

273 warnings.warn( 

274 "'ERROR_404_HELP' config setting is deprecated and will be " 

275 "removed in the future. Use 'RESTX_ERROR_404_HELP' instead.", 

276 DeprecationWarning, 

277 ) 

278 

279 def __getattr__(self, name): 

280 try: 

281 return getattr(self.default_namespace, name) 

282 except AttributeError: 

283 raise AttributeError("Api does not have {0} attribute".format(name)) 

284 

285 def _complete_url(self, url_part, registration_prefix): 

286 """ 

287 This method is used to defer the construction of the final url in 

288 the case that the Api is created with a Blueprint. 

289 

290 :param url_part: The part of the url the endpoint is registered with 

291 :param registration_prefix: The part of the url contributed by the 

292 blueprint. Generally speaking, BlueprintSetupState.url_prefix 

293 """ 

294 parts = (registration_prefix, self.prefix, url_part) 

295 return "".join(part for part in parts if part) 

296 

297 def _register_apidoc(self, app): 

298 conf = app.extensions.setdefault("restx", {}) 

299 if not conf.get("apidoc_registered", False): 

300 app.register_blueprint(apidoc.apidoc) 

301 conf["apidoc_registered"] = True 

302 

303 def _register_specs(self, app_or_blueprint): 

304 if self._add_specs: 

305 endpoint = str("specs") 

306 self._register_view( 

307 app_or_blueprint, 

308 SwaggerView, 

309 self.default_namespace, 

310 "/" + self.default_swagger_filename, 

311 endpoint=endpoint, 

312 resource_class_args=(self,), 

313 ) 

314 self.endpoints.add(endpoint) 

315 

316 def _register_doc(self, app_or_blueprint): 

317 if self._add_specs and self._doc: 

318 # Register documentation before root if enabled 

319 app_or_blueprint.add_url_rule(self._doc, "doc", self.render_doc) 

320 app_or_blueprint.add_url_rule(self.prefix or "/", "root", self.render_root) 

321 

322 def register_resource(self, namespace, resource, *urls, **kwargs): 

323 endpoint = kwargs.pop("endpoint", None) 

324 endpoint = str(endpoint or self.default_endpoint(resource, namespace)) 

325 

326 kwargs["endpoint"] = endpoint 

327 self.endpoints.add(endpoint) 

328 

329 if self.app is not None: 

330 self._register_view(self.app, resource, namespace, *urls, **kwargs) 

331 else: 

332 self.resources.append((resource, namespace, urls, kwargs)) 

333 return endpoint 

334 

335 def _configure_namespace_logger(self, app, namespace): 

336 for handler in app.logger.handlers: 

337 namespace.logger.addHandler(handler) 

338 namespace.logger.setLevel(app.logger.level) 

339 

340 def _register_view(self, app, resource, namespace, *urls, **kwargs): 

341 endpoint = kwargs.pop("endpoint", None) or camel_to_dash(resource.__name__) 

342 resource_class_args = kwargs.pop("resource_class_args", ()) 

343 resource_class_kwargs = kwargs.pop("resource_class_kwargs", {}) 

344 

345 # NOTE: 'view_functions' is cleaned up from Blueprint class in Flask 1.0 

346 if endpoint in getattr(app, "view_functions", {}): 

347 previous_view_class = app.view_functions[endpoint].__dict__["view_class"] 

348 

349 # if you override the endpoint with a different class, avoid the 

350 # collision by raising an exception 

351 if previous_view_class != resource: 

352 msg = "This endpoint (%s) is already set to the class %s." 

353 raise ValueError(msg % (endpoint, previous_view_class.__name__)) 

354 

355 resource.mediatypes = self.mediatypes_method() # Hacky 

356 resource.endpoint = endpoint 

357 

358 resource_func = self.output( 

359 resource.as_view( 

360 endpoint, self, *resource_class_args, **resource_class_kwargs 

361 ) 

362 ) 

363 

364 # Apply Namespace and Api decorators to a resource 

365 for decorator in chain(namespace.decorators, self.decorators): 

366 resource_func = decorator(resource_func) 

367 

368 for url in urls: 

369 # If this Api has a blueprint 

370 if self.blueprint: 

371 # And this Api has been setup 

372 if self.blueprint_setup: 

373 # Set the rule to a string directly, as the blueprint is already 

374 # set up. 

375 self.blueprint_setup.add_url_rule( 

376 url, view_func=resource_func, **kwargs 

377 ) 

378 continue 

379 else: 

380 # Set the rule to a function that expects the blueprint prefix 

381 # to construct the final url. Allows deferment of url finalization 

382 # in the case that the associated Blueprint has not yet been 

383 # registered to an application, so we can wait for the registration 

384 # prefix 

385 rule = partial(self._complete_url, url) 

386 else: 

387 # If we've got no Blueprint, just build a url with no prefix 

388 rule = self._complete_url(url, "") 

389 # Add the url to the application or blueprint 

390 app.add_url_rule(rule, view_func=resource_func, **kwargs) 

391 

392 def output(self, resource): 

393 """ 

394 Wraps a resource (as a flask view function), 

395 for cases where the resource does not directly return a response object 

396 

397 :param resource: The resource as a flask view function 

398 """ 

399 

400 @wraps(resource) 

401 def wrapper(*args, **kwargs): 

402 resp = resource(*args, **kwargs) 

403 if isinstance(resp, BaseResponse): 

404 return resp 

405 data, code, headers = unpack(resp) 

406 return self.make_response(data, code, headers=headers) 

407 

408 return wrapper 

409 

410 def make_response(self, data, *args, **kwargs): 

411 """ 

412 Looks up the representation transformer for the requested media 

413 type, invoking the transformer to create a response object. This 

414 defaults to default_mediatype if no transformer is found for the 

415 requested mediatype. If default_mediatype is None, a 406 Not 

416 Acceptable response will be sent as per RFC 2616 section 14.1 

417 

418 :param data: Python object containing response data to be transformed 

419 """ 

420 default_mediatype = ( 

421 kwargs.pop("fallback_mediatype", None) or self.default_mediatype 

422 ) 

423 mediatype = request.accept_mimetypes.best_match( 

424 self.representations, 

425 default=default_mediatype, 

426 ) 

427 if mediatype is None: 

428 raise NotAcceptable() 

429 if mediatype in self.representations: 

430 resp = self.representations[mediatype](data, *args, **kwargs) 

431 resp.headers["Content-Type"] = mediatype 

432 return resp 

433 elif mediatype == "text/plain": 

434 resp = original_flask_make_response(str(data), *args, **kwargs) 

435 resp.headers["Content-Type"] = "text/plain" 

436 return resp 

437 else: 

438 raise InternalServerError() 

439 

440 def documentation(self, func): 

441 """A decorator to specify a view function for the documentation""" 

442 self._doc_view = func 

443 return func 

444 

445 def render_root(self): 

446 self.abort(HTTPStatus.NOT_FOUND) 

447 

448 def render_doc(self): 

449 """Override this method to customize the documentation page""" 

450 if self._doc_view: 

451 return self._doc_view() 

452 elif not self._doc: 

453 self.abort(HTTPStatus.NOT_FOUND) 

454 return apidoc.ui_for(self) 

455 

456 def default_endpoint(self, resource, namespace): 

457 """ 

458 Provide a default endpoint for a resource on a given namespace. 

459 

460 Endpoints are ensured not to collide. 

461 

462 Override this method specify a custom algorithm for default endpoint. 

463 

464 :param Resource resource: the resource for which we want an endpoint 

465 :param Namespace namespace: the namespace holding the resource 

466 :returns str: An endpoint name 

467 """ 

468 endpoint = camel_to_dash(resource.__name__) 

469 if namespace is not self.default_namespace: 

470 endpoint = "{ns.name}_{endpoint}".format(ns=namespace, endpoint=endpoint) 

471 if endpoint in self.endpoints: 

472 suffix = 2 

473 while True: 

474 new_endpoint = "{base}_{suffix}".format(base=endpoint, suffix=suffix) 

475 if new_endpoint not in self.endpoints: 

476 endpoint = new_endpoint 

477 break 

478 suffix += 1 

479 return endpoint 

480 

481 def get_ns_path(self, ns): 

482 return self.ns_paths.get(ns) 

483 

484 def ns_urls(self, ns, urls): 

485 path = self.get_ns_path(ns) or ns.path 

486 return [path + url for url in urls] 

487 

488 def add_namespace(self, ns, path=None): 

489 """ 

490 This method registers resources from namespace for current instance of api. 

491 You can use argument path for definition custom prefix url for namespace. 

492 

493 :param Namespace ns: the namespace 

494 :param path: registration prefix of namespace 

495 """ 

496 if ns not in self.namespaces: 

497 self.namespaces.append(ns) 

498 if self not in ns.apis: 

499 ns.apis.append(self) 

500 # Associate ns with prefix-path 

501 if path is not None: 

502 self.ns_paths[ns] = path 

503 # Register resources 

504 for r in ns.resources: 

505 urls = self.ns_urls(ns, r.urls) 

506 self.register_resource(ns, r.resource, *urls, **r.kwargs) 

507 # Register models 

508 for name, definition in ns.models.items(): 

509 self.models[name] = definition 

510 if not self.blueprint and self.app is not None: 

511 self._configure_namespace_logger(self.app, ns) 

512 

513 def namespace(self, *args, **kwargs): 

514 """ 

515 A namespace factory. 

516 

517 :returns Namespace: a new namespace instance 

518 """ 

519 kwargs["ordered"] = kwargs.get("ordered", self.ordered) 

520 ns = Namespace(*args, **kwargs) 

521 self.add_namespace(ns) 

522 return ns 

523 

524 def endpoint(self, name): 

525 if self.blueprint: 

526 return "{0}.{1}".format(self.blueprint.name, name) 

527 else: 

528 return name 

529 

530 @property 

531 def specs_url(self): 

532 """ 

533 The Swagger specifications relative url (ie. `swagger.json`). If 

534 the spec_url_scheme attribute is set, then the full url is provided instead 

535 (e.g. http://localhost/swaggger.json). 

536 

537 :rtype: str 

538 """ 

539 external = None if self.url_scheme is None else True 

540 return url_for( 

541 self.endpoint("specs"), _scheme=self.url_scheme, _external=external 

542 ) 

543 

544 @property 

545 def base_url(self): 

546 """ 

547 The API base absolute url 

548 

549 :rtype: str 

550 """ 

551 return url_for(self.endpoint("root"), _scheme=self.url_scheme, _external=True) 

552 

553 @property 

554 def base_path(self): 

555 """ 

556 The API path 

557 

558 :rtype: str 

559 """ 

560 return url_for(self.endpoint("root"), _external=False) 

561 

562 @cached_property 

563 def __schema__(self): 

564 """ 

565 The Swagger specifications/schema for this API 

566 

567 :returns dict: the schema as a serializable dict 

568 """ 

569 if not self._schema: 

570 try: 

571 self._schema = Swagger(self).as_dict() 

572 except Exception: 

573 # Log the source exception for debugging purpose 

574 # and return an error message 

575 msg = "Unable to render schema" 

576 log.exception(msg) # This will provide a full traceback 

577 return {"error": msg} 

578 return self._schema 

579 

580 @property 

581 def _own_and_child_error_handlers(self): 

582 rv = OrderedDict() 

583 rv.update(self.error_handlers) 

584 for ns in self.namespaces: 

585 for exception, handler in ns.error_handlers.items(): 

586 rv[exception] = handler 

587 return rv 

588 

589 def errorhandler(self, exception): 

590 """A decorator to register an error handler for a given exception""" 

591 if inspect.isclass(exception) and issubclass(exception, Exception): 

592 # Register an error handler for a given exception 

593 def wrapper(func): 

594 self.error_handlers[exception] = func 

595 return func 

596 

597 return wrapper 

598 else: 

599 # Register the default error handler 

600 self._default_error_handler = exception 

601 return exception 

602 

603 def owns_endpoint(self, endpoint): 

604 """ 

605 Tests if an endpoint name (not path) belongs to this Api. 

606 Takes into account the Blueprint name part of the endpoint name. 

607 

608 :param str endpoint: The name of the endpoint being checked 

609 :return: bool 

610 """ 

611 

612 if self.blueprint: 

613 if endpoint.startswith(self.blueprint.name): 

614 endpoint = endpoint.split(self.blueprint.name + ".", 1)[-1] 

615 else: 

616 return False 

617 return endpoint in self.endpoints 

618 

619 def _should_use_fr_error_handler(self): 

620 """ 

621 Determine if error should be handled with FR or default Flask 

622 

623 The goal is to return Flask error handlers for non-FR-related routes, 

624 and FR errors (with the correct media type) for FR endpoints. This 

625 method currently handles 404 and 405 errors. 

626 

627 :return: bool 

628 """ 

629 adapter = current_app.create_url_adapter(request) 

630 

631 try: 

632 adapter.match() 

633 except MethodNotAllowed as e: 

634 # Check if the other HTTP methods at this url would hit the Api 

635 valid_route_method = e.valid_methods[0] 

636 rule, _ = adapter.match(method=valid_route_method, return_rule=True) 

637 return self.owns_endpoint(rule.endpoint) 

638 except NotFound: 

639 return self.catch_all_404s 

640 except Exception: 

641 # Werkzeug throws other kinds of exceptions, such as Redirect 

642 pass 

643 

644 def _has_fr_route(self): 

645 """Encapsulating the rules for whether the request was to a Flask endpoint""" 

646 # 404's, 405's, which might not have a url_rule 

647 if self._should_use_fr_error_handler(): 

648 return True 

649 # for all other errors, just check if FR dispatched the route 

650 if not request.url_rule: 

651 return False 

652 return self.owns_endpoint(request.url_rule.endpoint) 

653 

654 def error_router(self, original_handler, e): 

655 """ 

656 This function decides whether the error occurred in a flask-restx 

657 endpoint or not. If it happened in a flask-restx endpoint, our 

658 handler will be dispatched. If it happened in an unrelated view, the 

659 app's original error handler will be dispatched. 

660 In the event that the error occurred in a flask-restx endpoint but 

661 the local handler can't resolve the situation, the router will fall 

662 back onto the original_handler as last resort. 

663 

664 :param function original_handler: the original Flask error handler for the app 

665 :param Exception e: the exception raised while handling the request 

666 """ 

667 if self._has_fr_route(): 

668 try: 

669 return self.handle_error(e) 

670 except Exception as f: 

671 return original_handler(f) 

672 return original_handler(e) 

673 

674 def _propagate_exceptions(self): 

675 """ 

676 Returns the value of the ``PROPAGATE_EXCEPTIONS`` configuration 

677 value in case it's set, otherwise return true if app.debug or 

678 app.testing is set. This method was deprecated in Flask 2.3 but 

679 we still need it for our error handlers. 

680 """ 

681 rv = current_app.config.get("PROPAGATE_EXCEPTIONS") 

682 if rv is not None: 

683 return rv 

684 return current_app.testing or current_app.debug 

685 

686 def handle_error(self, e): 

687 """ 

688 Error handler for the API transforms a raised exception into a Flask response, 

689 with the appropriate HTTP status code and body. 

690 

691 :param Exception e: the raised Exception object 

692 

693 """ 

694 # When propagate_exceptions is set, do not return the exception to the 

695 # client if a handler is configured for the exception. 

696 if ( 

697 not isinstance(e, HTTPException) 

698 and self._propagate_exceptions() 

699 and not isinstance(e, tuple(self._own_and_child_error_handlers.keys())) 

700 ): 

701 exc_type, exc_value, tb = sys.exc_info() 

702 if exc_value is e: 

703 raise 

704 else: 

705 raise e 

706 

707 include_message_in_response = current_app.config.get( 

708 "ERROR_INCLUDE_MESSAGE", True 

709 ) 

710 default_data = {} 

711 

712 headers = Headers() 

713 

714 for typecheck, handler in self._own_and_child_error_handlers.items(): 

715 if isinstance(e, typecheck): 

716 result = handler(e) 

717 default_data, code, headers = unpack( 

718 result, HTTPStatus.INTERNAL_SERVER_ERROR 

719 ) 

720 break 

721 else: 

722 # Flask docs say: "This signal is not sent for HTTPException or other exceptions that have error handlers 

723 # registered, unless the exception was raised from an error handler." 

724 got_request_exception.send(current_app._get_current_object(), exception=e) 

725 

726 if isinstance(e, HTTPException): 

727 code = None 

728 if e.code is not None: 

729 code = HTTPStatus(e.code) 

730 elif e.response is not None: 

731 code = HTTPStatus(e.response.status_code) 

732 if include_message_in_response: 

733 default_data = {"message": e.description or code.phrase} 

734 headers = e.get_response().headers 

735 elif self._default_error_handler: 

736 result = self._default_error_handler(e) 

737 default_data, code, headers = unpack( 

738 result, HTTPStatus.INTERNAL_SERVER_ERROR 

739 ) 

740 else: 

741 code = HTTPStatus.INTERNAL_SERVER_ERROR 

742 if include_message_in_response: 

743 default_data = { 

744 "message": code.phrase, 

745 } 

746 

747 if include_message_in_response: 

748 default_data["message"] = default_data.get("message", str(e)) 

749 

750 data = getattr(e, "data", default_data) 

751 fallback_mediatype = None 

752 

753 if code >= HTTPStatus.INTERNAL_SERVER_ERROR: 

754 exc_info = sys.exc_info() 

755 if exc_info[1] is None: 

756 exc_info = None 

757 current_app.log_exception(exc_info) 

758 

759 elif ( 

760 code == HTTPStatus.NOT_FOUND 

761 and current_app.config.get("RESTX_ERROR_404_HELP", True) 

762 and include_message_in_response 

763 ): 

764 data["message"] = self._help_on_404(data.get("message", None)) 

765 

766 elif code == HTTPStatus.NOT_ACCEPTABLE and self.default_mediatype is None: 

767 # if we are handling NotAcceptable (406), make sure that 

768 # make_response uses a representation we support as the 

769 # default mediatype (so that make_response doesn't throw 

770 # another NotAcceptable error). 

771 supported_mediatypes = list(self.representations.keys()) 

772 fallback_mediatype = ( 

773 supported_mediatypes[0] if supported_mediatypes else "text/plain" 

774 ) 

775 

776 # Remove blacklisted headers 

777 for header in HEADERS_BLACKLIST: 

778 headers.pop(header, None) 

779 

780 resp = self.make_response( 

781 data, code, headers, fallback_mediatype=fallback_mediatype 

782 ) 

783 

784 if code == HTTPStatus.UNAUTHORIZED: 

785 resp = self.unauthorized(resp) 

786 return resp 

787 

788 def _help_on_404(self, message=None): 

789 rules = dict( 

790 [ 

791 (RE_RULES.sub("", rule.rule), rule.rule) 

792 for rule in current_app.url_map.iter_rules() 

793 ] 

794 ) 

795 close_matches = difflib.get_close_matches(request.path, rules.keys()) 

796 if close_matches: 

797 # If we already have a message, add punctuation and continue it. 

798 message = "".join( 

799 ( 

800 (message.rstrip(".") + ". ") if message else "", 

801 "You have requested this URI [", 

802 request.path, 

803 "] but did you mean ", 

804 " or ".join((rules[match] for match in close_matches)), 

805 " ?", 

806 ) 

807 ) 

808 return message 

809 

810 def as_postman(self, urlvars=False, swagger=False): 

811 """ 

812 Serialize the API as Postman collection (v1) 

813 

814 :param bool urlvars: whether to include or not placeholders for query strings 

815 :param bool swagger: whether to include or not the swagger.json specifications 

816 

817 """ 

818 return PostmanCollectionV1(self, swagger=swagger).as_dict(urlvars=urlvars) 

819 

820 @property 

821 def payload(self): 

822 """Store the input payload in the current request context""" 

823 return request.get_json() 

824 

825 @property 

826 def refresolver(self): 

827 if not self._refresolver: 

828 self._refresolver = RefResolver.from_schema(self.__schema__) 

829 return self._refresolver 

830 

831 @staticmethod 

832 def _blueprint_setup_add_url_rule_patch( 

833 blueprint_setup, rule, endpoint=None, view_func=None, **options 

834 ): 

835 """ 

836 Method used to patch BlueprintSetupState.add_url_rule for setup 

837 state instance corresponding to this Api instance. Exists primarily 

838 to enable _complete_url's function. 

839 

840 :param blueprint_setup: The BlueprintSetupState instance (self) 

841 :param rule: A string or callable that takes a string and returns a 

842 string(_complete_url) that is the url rule for the endpoint 

843 being registered 

844 :param endpoint: See BlueprintSetupState.add_url_rule 

845 :param view_func: See BlueprintSetupState.add_url_rule 

846 :param **options: See BlueprintSetupState.add_url_rule 

847 """ 

848 

849 if callable(rule): 

850 rule = rule(blueprint_setup.url_prefix) 

851 elif blueprint_setup.url_prefix: 

852 rule = blueprint_setup.url_prefix + rule 

853 options.setdefault("subdomain", blueprint_setup.subdomain) 

854 if endpoint is None: 

855 endpoint = endpoint_from_view_func(view_func) 

856 defaults = blueprint_setup.url_defaults 

857 if "defaults" in options: 

858 defaults = dict(defaults, **options.pop("defaults")) 

859 blueprint_setup.app.add_url_rule( 

860 rule, 

861 "%s.%s" % (blueprint_setup.blueprint.name, endpoint), 

862 view_func, 

863 defaults=defaults, 

864 **options 

865 ) 

866 

867 def _deferred_blueprint_init(self, setup_state): 

868 """ 

869 Synchronize prefix between blueprint/api and registration options, then 

870 perform initialization with setup_state.app :class:`flask.Flask` object. 

871 When a :class:`flask_restx.Api` object is initialized with a blueprint, 

872 this method is recorded on the blueprint to be run when the blueprint is later 

873 registered to a :class:`flask.Flask` object. This method also monkeypatches 

874 BlueprintSetupState.add_url_rule with _blueprint_setup_add_url_rule_patch. 

875 

876 :param setup_state: The setup state object passed to deferred functions 

877 during blueprint registration 

878 :type setup_state: flask.blueprints.BlueprintSetupState 

879