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

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 

17try: 

18 from flask.helpers import _endpoint_from_view_func 

19except ImportError: 

20 from flask.scaffold import _endpoint_from_view_func 

21from flask.signals import got_request_exception 

22 

23from jsonschema import RefResolver 

24 

25from werkzeug.utils import cached_property 

26from werkzeug.datastructures import Headers 

27from werkzeug.exceptions import ( 

28 HTTPException, 

29 MethodNotAllowed, 

30 NotFound, 

31 NotAcceptable, 

32 InternalServerError, 

33) 

34 

35from werkzeug import __version__ as werkzeug_version 

36 

37if werkzeug_version.split(".")[0] >= "2": 

38 from werkzeug.wrappers import Response as BaseResponse 

39else: 

40 from werkzeug.wrappers import BaseResponse 

41 

42from . import apidoc 

43from .mask import ParseError, MaskError 

44from .namespace import Namespace 

45from .postman import PostmanCollectionV1 

46from .resource import Resource 

47from .swagger import Swagger 

48from .utils import default_id, camel_to_dash, unpack 

49from .representations import output_json 

50from ._http import HTTPStatus 

51 

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

53 

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

55HEADERS_BLACKLIST = ("Content-Length",) 

56 

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

58 

59log = logging.getLogger(__name__) 

60 

61 

62class Api(object): 

63 """ 

64 The main entry point for the application. 

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

66 

67 >>> app = Flask(__name__) 

68 >>> api = Api(app) 

69 

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

71 after it has been constructed. 

72 

73 The endpoint parameter prefix all views and resources: 

74 

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

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

77 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

93 (Default to '/') 

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

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

96 to handle 404 errors throughout your app 

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

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

99 responses (default 'False') 

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

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

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

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

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

105 proxy. 

106 :param str default_swagger_filename: The default swagger filename. 

107 """ 

108 

109 def __init__( 

110 self, 

111 app=None, 

112 version="1.0", 

113 title=None, 

114 description=None, 

115 terms_url=None, 

116 license=None, 

117 license_url=None, 

118 contact=None, 

119 contact_url=None, 

120 contact_email=None, 

121 authorizations=None, 

122 security=None, 

123 doc="/", 

124 default_id=default_id, 

125 default="default", 

126 default_label="Default namespace", 

127 validate=None, 

128 tags=None, 

129 prefix="", 

130 ordered=False, 

131 default_mediatype="application/json", 

132 decorators=None, 

133 catch_all_404s=False, 

134 serve_challenge_on_401=False, 

135 format_checker=None, 

136 url_scheme=None, 

137 default_swagger_filename="swagger.json", 

138 **kwargs 

139 ): 

140 self.version = version 

141 self.title = title or "API" 

142 self.description = description 

143 self.terms_url = terms_url 

144 self.contact = contact 

145 self.contact_email = contact_email 

146 self.contact_url = contact_url 

147 self.license = license 

148 self.license_url = license_url 

149 self.authorizations = authorizations 

150 self.security = security 

151 self.default_id = default_id 

152 self.ordered = ordered 

153 self._validate = validate 

154 self._doc = doc 

155 self._doc_view = None 

156 self._default_error_handler = None 

157 self.tags = tags or [] 

158 

159 self.error_handlers = OrderedDict( 

160 { 

161 ParseError: mask_parse_error_handler, 

162 MaskError: mask_error_handler, 

163 } 

164 ) 

165 self._schema = None 

166 self.models = {} 

167 self._refresolver = None 

168 self.format_checker = format_checker 

169 self.namespaces = [] 

170 self.default_swagger_filename = default_swagger_filename 

171 

172 self.ns_paths = dict() 

173 

174 self.representations = OrderedDict(DEFAULT_REPRESENTATIONS) 

175 self.urls = {} 

176 self.prefix = prefix 

177 self.default_mediatype = default_mediatype 

178 self.decorators = decorators if decorators else [] 

179 self.catch_all_404s = catch_all_404s 

180 self.serve_challenge_on_401 = serve_challenge_on_401 

181 self.blueprint_setup = None 

182 self.endpoints = set() 

183 self.resources = [] 

184 self.app = None 

185 self.blueprint = None 

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

187 # in self._configure_namespace_logger 

188 self.default_namespace = self.namespace( 

189 default, 

190 default_label, 

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

192 validate=validate, 

193 api=self, 

194 path="/", 

195 ) 

196 self.url_scheme = url_scheme 

197 if app is not None: 

198 self.app = app 

199 self.init_app(app) 

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

201 

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

203 """ 

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

205 

206 >>> app = Flask(__name__) 

207 >>> api = Api() 

208 >>> api.init_app(app) 

209 

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

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

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

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

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

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

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

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

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

219 reverse proxy. 

220 """ 

221 self.app = app 

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

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

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

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

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

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

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

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

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

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

232 self._register_specs(app) 

233 self._register_doc(app) 

234 

235 # If app is a blueprint, defer the initialization 

236 try: 

237 app.record(self._deferred_blueprint_init) 

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

239 except AttributeError: 

240 self._init_app(app) 

241 else: 

242 self.blueprint = app 

243 

244 def _init_app(self, app): 

245 """ 

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

247 

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

249 """ 

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

251 app.handle_user_exception = partial( 

252 self.error_router, app.handle_user_exception 

253 ) 

254 

255 if len(self.resources) > 0: 

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

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

258 

259 for ns in self.namespaces: 

260 self._configure_namespace_logger(app, ns) 

261 

262 self._register_apidoc(app) 

263 self._validate = ( 

264 self._validate 

265 if self._validate is not None 

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

267 ) 

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

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

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

271 

272 # check for deprecated config variable names 

273 if "ERROR_404_HELP" in app.config: 

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

275 warnings.warn( 

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

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

278 DeprecationWarning, 

279 ) 

280 

281 def __getattr__(self, name): 

282 try: 

283 return getattr(self.default_namespace, name) 

284 except AttributeError: 

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

286 

287 def _complete_url(self, url_part, registration_prefix): 

288 """ 

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

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

291 

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

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

294 blueprint. Generally speaking, BlueprintSetupState.url_prefix 

295 """ 

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

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

298 

299 def _register_apidoc(self, app): 

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

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

302 app.register_blueprint(apidoc.apidoc) 

303 conf["apidoc_registered"] = True 

304 

305 def _register_specs(self, app_or_blueprint): 

306 if self._add_specs: 

307 endpoint = str("specs") 

308 self._register_view( 

309 app_or_blueprint, 

310 SwaggerView, 

311 self.default_namespace, 

312 "/" + self.default_swagger_filename, 

313 endpoint=endpoint, 

314 resource_class_args=(self,), 

315 ) 

316 self.endpoints.add(endpoint) 

317 

318 def _register_doc(self, app_or_blueprint): 

319 if self._add_specs and self._doc: 

320 # Register documentation before root if enabled 

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

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

323 

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

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

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

327 

328 kwargs["endpoint"] = endpoint 

329 self.endpoints.add(endpoint) 

330 

331 if self.app is not None: 

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

333 else: 

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

335 return endpoint 

336 

337 def _configure_namespace_logger(self, app, namespace): 

338 for handler in app.logger.handlers: 

339 namespace.logger.addHandler(handler) 

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

341 

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

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

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

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

346 

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

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

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

350 

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

352 # collision by raising an exception 

353 if previous_view_class != resource: 

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

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

356 

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

358 resource.endpoint = endpoint 

359 

360 resource_func = self.output( 

361 resource.as_view( 

362 endpoint, self, *resource_class_args, **resource_class_kwargs 

363 ) 

364 ) 

365 

366 # Apply Namespace and Api decorators to a resource 

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

368 resource_func = decorator(resource_func) 

369 

370 for url in urls: 

371 # If this Api has a blueprint 

372 if self.blueprint: 

373 # And this Api has been setup 

374 if self.blueprint_setup: 

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

376 # set up. 

377 self.blueprint_setup.add_url_rule( 

378 url, view_func=resource_func, **kwargs 

379 ) 

380 continue 

381 else: 

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

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

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

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

386 # prefix 

387 rule = partial(self._complete_url, url) 

388 else: 

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

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

391 # Add the url to the application or blueprint 

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

393 

394 def output(self, resource): 

395 """ 

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

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

398 

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

400 """ 

401 

402 @wraps(resource) 

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

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

405 if isinstance(resp, BaseResponse): 

406 return resp 

407 data, code, headers = unpack(resp) 

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

409 

410 return wrapper 

411 

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

413 """ 

414 Looks up the representation transformer for the requested media 

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

416 defaults to default_mediatype if no transformer is found for the 

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

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

419 

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

421 """ 

422 default_mediatype = ( 

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

424 ) 

425 mediatype = request.accept_mimetypes.best_match( 

426 self.representations, 

427 default=default_mediatype, 

428 ) 

429 if mediatype is None: 

430 raise NotAcceptable() 

431 if mediatype in self.representations: 

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

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

434 return resp 

435 elif mediatype == "text/plain": 

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

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

438 return resp 

439 else: 

440 raise InternalServerError() 

441 

442 def documentation(self, func): 

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

444 self._doc_view = func 

445 return func 

446 

447 def render_root(self): 

448 self.abort(HTTPStatus.NOT_FOUND) 

449 

450 def render_doc(self): 

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

452 if self._doc_view: 

453 return self._doc_view() 

454 elif not self._doc: 

455 self.abort(HTTPStatus.NOT_FOUND) 

456 return apidoc.ui_for(self) 

457 

458 def default_endpoint(self, resource, namespace): 

459 """ 

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

461 

462 Endpoints are ensured not to collide. 

463 

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

465 

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

467 :param Namespace namespace: the namespace holding the resource 

468 :returns str: An endpoint name 

469 """ 

470 endpoint = camel_to_dash(resource.__name__) 

471 if namespace is not self.default_namespace: 

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

473 if endpoint in self.endpoints: 

474 suffix = 2 

475 while True: 

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

477 if new_endpoint not in self.endpoints: 

478 endpoint = new_endpoint 

479 break 

480 suffix += 1 

481 return endpoint 

482 

483 def get_ns_path(self, ns): 

484 return self.ns_paths.get(ns) 

485 

486 def ns_urls(self, ns, urls): 

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

488 return [path + url for url in urls] 

489 

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

491 """ 

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

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

494 

495 :param Namespace ns: the namespace 

496 :param path: registration prefix of namespace 

497 """ 

498 if ns not in self.namespaces: 

499 self.namespaces.append(ns) 

500 if self not in ns.apis: 

501 ns.apis.append(self) 

502 # Associate ns with prefix-path 

503 if path is not None: 

504 self.ns_paths[ns] = path 

505 # Register resources 

506 for r in ns.resources: 

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

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

509 # Register models 

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

511 self.models[name] = definition 

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

513 self._configure_namespace_logger(self.app, ns) 

514 

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

516 """ 

517 A namespace factory. 

518 

519 :returns Namespace: a new namespace instance 

520 """ 

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

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

523 self.add_namespace(ns) 

524 return ns 

525 

526 def endpoint(self, name): 

527 if self.blueprint: 

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

529 else: 

530 return name 

531 

532 @property 

533 def specs_url(self): 

534 """ 

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

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

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

538 

539 :rtype: str 

540 """ 

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

542 return url_for( 

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

544 ) 

545 

546 @property 

547 def base_url(self): 

548 """ 

549 The API base absolute url 

550 

551 :rtype: str 

552 """ 

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

554 

555 @property 

556 def base_path(self): 

557 """ 

558 The API path 

559 

560 :rtype: str 

561 """ 

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

563 

564 @cached_property 

565 def __schema__(self): 

566 """ 

567 The Swagger specifications/schema for this API 

568 

569 :returns dict: the schema as a serializable dict 

570 """ 

571 if not self._schema: 

572 try: 

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

574 except Exception: 

575 # Log the source exception for debugging purpose 

576 # and return an error message 

577 msg = "Unable to render schema" 

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

579 return {"error": msg} 

580 return self._schema 

581 

582 @property 

583 def _own_and_child_error_handlers(self): 

584 rv = OrderedDict() 

585 rv.update(self.error_handlers) 

586 for ns in self.namespaces: 

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

588 rv[exception] = handler 

589 return rv 

590 

591 def errorhandler(self, exception): 

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

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

594 # Register an error handler for a given exception 

595 def wrapper(func): 

596 self.error_handlers[exception] = func 

597 return func 

598 

599 return wrapper 

600 else: 

601 # Register the default error handler 

602 self._default_error_handler = exception 

603 return exception 

604 

605 def owns_endpoint(self, endpoint): 

606 """ 

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

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

609 

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

611 :return: bool 

612 """ 

613 

614 if self.blueprint: 

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

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

617 else: 

618 return False 

619 return endpoint in self.endpoints 

620 

621 def _should_use_fr_error_handler(self): 

622 """ 

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

624 

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

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

627 method currently handles 404 and 405 errors. 

628 

629 :return: bool 

630 """ 

631 adapter = current_app.create_url_adapter(request) 

632 

633 try: 

634 adapter.match() 

635 except MethodNotAllowed as e: 

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

637 valid_route_method = e.valid_methods[0] 

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

639 return self.owns_endpoint(rule.endpoint) 

640 except NotFound: 

641 return self.catch_all_404s 

642 except Exception: 

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

644 pass 

645 

646 def _has_fr_route(self): 

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

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

649 if self._should_use_fr_error_handler(): 

650 return True 

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

652 if not request.url_rule: 

653 return False 

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

655 

656 def error_router(self, original_handler, e): 

657 """ 

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

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

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

661 app's original error handler will be dispatched. 

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

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

664 back onto the original_handler as last resort. 

665 

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

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

668 """ 

669 if self._has_fr_route(): 

670 try: 

671 return self.handle_error(e) 

672 except Exception as f: 

673 return original_handler(f) 

674 return original_handler(e) 

675 

676 def _propagate_exceptions(self): 

677 """ 

678 Returns the value of the ``PROPAGATE_EXCEPTIONS`` configuration 

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

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

681 we still need it for our error handlers. 

682 """ 

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

684 if rv is not None: 

685 return rv 

686 return current_app.testing or current_app.debug 

687 

688 def handle_error(self, e): 

689 """ 

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

691 with the appropriate HTTP status code and body. 

692 

693 :param Exception e: the raised Exception object 

694 

695 """ 

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

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

698 if ( 

699 not isinstance(e, HTTPException) 

700 and self._propagate_exceptions() 

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

702 ): 

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

704 if exc_value is e: 

705 raise 

706 else: 

707 raise e 

708 

709 include_message_in_response = current_app.config.get( 

710 "ERROR_INCLUDE_MESSAGE", True 

711 ) 

712 default_data = {} 

713 

714 headers = Headers() 

715 

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

717 if isinstance(e, typecheck): 

718 result = handler(e) 

719 default_data, code, headers = unpack( 

720 result, HTTPStatus.INTERNAL_SERVER_ERROR 

721 ) 

722 break 

723 else: 

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

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

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

727 

728 if isinstance(e, HTTPException): 

729 code = HTTPStatus(e.code) 

730 if include_message_in_response: 

731 default_data = {"message": getattr(e, "description", code.phrase)} 

732 headers = e.get_response().headers 

733 elif self._default_error_handler: 

734 result = self._default_error_handler(e) 

735 default_data, code, headers = unpack( 

736 result, HTTPStatus.INTERNAL_SERVER_ERROR 

737 ) 

738 else: 

739 code = HTTPStatus.INTERNAL_SERVER_ERROR 

740 if include_message_in_response: 

741 default_data = { 

742 "message": code.phrase, 

743 } 

744 

745 if include_message_in_response: 

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

747 

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

749 fallback_mediatype = None 

750 

751 if code >= HTTPStatus.INTERNAL_SERVER_ERROR: 

752 exc_info = sys.exc_info() 

753 if exc_info[1] is None: 

754 exc_info = None 

755 current_app.log_exception(exc_info) 

756 

757 elif ( 

758 code == HTTPStatus.NOT_FOUND 

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

760 and include_message_in_response 

761 ): 

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

763 

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

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

766 # make_response uses a representation we support as the 

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

768 # another NotAcceptable error). 

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

770 fallback_mediatype = ( 

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

772 ) 

773 

774 # Remove blacklisted headers 

775 for header in HEADERS_BLACKLIST: 

776 headers.pop(header, None) 

777 

778 resp = self.make_response( 

779 data, code, headers, fallback_mediatype=fallback_mediatype 

780 ) 

781 

782 if code == HTTPStatus.UNAUTHORIZED: 

783 resp = self.unauthorized(resp) 

784 return resp 

785 

786 def _help_on_404(self, message=None): 

787 rules = dict( 

788 [ 

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

790 for rule in current_app.url_map.iter_rules() 

791 ] 

792 ) 

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

794 if close_matches: 

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

796 message = "".join( 

797 ( 

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

799 "You have requested this URI [", 

800 request.path, 

801 "] but did you mean ", 

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

803 " ?", 

804 ) 

805 ) 

806 return message 

807 

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

809 """ 

810 Serialize the API as Postman collection (v1) 

811 

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

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

814 

815 """ 

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

817 

818 @property 

819 def payload(self): 

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

821 return request.get_json() 

822 

823 @property 

824 def refresolver(self): 

825 if not self._refresolver: 

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

827 return self._refresolver 

828 

829 @staticmethod 

830 def _blueprint_setup_add_url_rule_patch( 

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

832 ): 

833 """ 

834 Method used to patch BlueprintSetupState.add_url_rule for setup 

835 state instance corresponding to this Api instance. Exists primarily 

836 to enable _complete_url's function. 

837 

838 :param blueprint_setup: The BlueprintSetupState instance (self) 

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

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

841 being registered 

842 :param endpoint: See BlueprintSetupState.add_url_rule 

843 :param view_func: See BlueprintSetupState.add_url_rule 

844 :param **options: See BlueprintSetupState.add_url_rule 

845 """ 

846 

847 if callable(rule): 

848 rule = rule(blueprint_setup.url_prefix) 

849 elif blueprint_setup.url_prefix: 

850 rule = blueprint_setup.url_prefix + rule 

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

852 if endpoint is None: 

853 endpoint = _endpoint_from_view_func(view_func) 

854 defaults = blueprint_setup.url_defaults 

855 if "defaults" in options: 

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

857 blueprint_setup.app.add_url_rule( 

858 rule, 

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

860 view_func, 

861 defaults=defaults, 

862 **options 

863 ) 

864 

865 def _deferred_blueprint_init(self, setup_state): 

866 """ 

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

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

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

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

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

872 BlueprintSetupState.add_url_rule with _blueprint_setup_add_url_rule_patch. 

873 

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

875 during blueprint registration 

876 :type setup_state: flask.blueprints.BlueprintSetupState