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

1import difflib 

2import inspect 

3from itertools import chain 

4import logging 

5import threading 

6import operator 

7import re 

8import sys 

9import warnings 

10 

11from collections import OrderedDict 

12from functools import wraps, partial 

13from types import MethodType 

14 

15from flask import url_for, request, current_app 

16from flask import make_response as original_flask_make_response 

17 

18from flask.signals import got_request_exception 

19 

20from referencing import Registry 

21 

22from werkzeug.utils import cached_property 

23from werkzeug.datastructures import Headers 

24from werkzeug.exceptions import ( 

25 HTTPException, 

26 MethodNotAllowed, 

27 NotFound, 

28 NotAcceptable, 

29 InternalServerError, 

30) 

31 

32from . import apidoc 

33from .mask import ParseError, MaskError 

34from .namespace import Namespace 

35from .postman import PostmanCollectionV1 

36from .resource import Resource 

37from .swagger import Swagger 

38from .utils import ( 

39 default_id, 

40 camel_to_dash, 

41 unpack, 

42 import_check_view_func, 

43 BaseResponse, 

44) 

45from .representations import output_json 

46from ._http import HTTPStatus 

47 

48endpoint_from_view_func = import_check_view_func() 

49 

50 

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

52 

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

54HEADERS_BLACKLIST = ("Content-Length",) 

55 

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

57 

58log = logging.getLogger(__name__) 

59 

60 

61class Api(object): 

62 """ 

63 The main entry point for the application. 

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

65 

66 >>> app = Flask(__name__) 

67 >>> api = Api(app) 

68 

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

70 after it has been constructed. 

71 

72 The endpoint parameter prefix all views and resources: 

73 

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

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

76 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

92 (Default to '/') 

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

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

95 to handle 404 errors throughout your app 

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

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

98 responses (default 'False') 

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

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

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

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

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

104 proxy. 

105 :param str default_swagger_filename: The default swagger filename. 

106 """ 

107 

108 def __init__( 

109 self, 

110 app=None, 

111 version="1.0", 

112 title=None, 

113 description=None, 

114 terms_url=None, 

115 license=None, 

116 license_url=None, 

117 contact=None, 

118 contact_url=None, 

119 contact_email=None, 

120 authorizations=None, 

121 security=None, 

122 doc="/", 

123 default_id=default_id, 

124 default="default", 

125 default_label="Default namespace", 

126 validate=None, 

127 tags=None, 

128 prefix="", 

129 ordered=False, 

130 default_mediatype="application/json", 

131 decorators=None, 

132 catch_all_404s=False, 

133 serve_challenge_on_401=False, 

134 format_checker=None, 

135 url_scheme=None, 

136 default_swagger_filename="swagger.json", 

137 **kwargs, 

138 ): 

139 self.version = version 

140 self.title = title or "API" 

141 self.description = description 

142 self.terms_url = terms_url 

143 self.contact = contact 

144 self.contact_email = contact_email 

145 self.contact_url = contact_url 

146 self.license = license 

147 self.license_url = license_url 

148 self.authorizations = authorizations 

149 self.security = security 

150 self.default_id = default_id 

151 self.ordered = ordered 

152 self._validate = validate 

153 self._doc = doc 

154 self._doc_view = None 

155 self._default_error_handler = None 

156 self.tags = tags or [] 

157 

158 self.error_handlers = OrderedDict( 

159 { 

160 ParseError: mask_parse_error_handler, 

161 MaskError: mask_error_handler, 

162 } 

163 ) 

164 self._schema = None 

165 self._schema_lock = threading.Lock() 

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 # Guard schema initialization to avoid concurrent construction on first access 

573 with self._schema_lock: 

574 if not self._schema: 

575 try: 

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

577 except Exception: 

578 # Log the source exception for debugging purpose 

579 # and return an error message 

580 msg = "Unable to render schema" 

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

582 return {"error": msg} 

583 return self._schema 

584 

585 @property 

586 def _own_and_child_error_handlers(self): 

587 rv = OrderedDict() 

588 rv.update(self.error_handlers) 

589 for ns in self.namespaces: 

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

591 rv[exception] = handler 

592 return rv 

593 

594 def errorhandler(self, exception): 

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

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

597 # Register an error handler for a given exception 

598 def wrapper(func): 

599 self.error_handlers[exception] = func 

600 return func 

601 

602 return wrapper 

603 else: 

604 # Register the default error handler 

605 self._default_error_handler = exception 

606 return exception 

607 

608 def owns_endpoint(self, endpoint): 

609 """ 

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

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

612 

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

614 :return: bool 

615 """ 

616 

617 if self.blueprint: 

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

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

620 else: 

621 return False 

622 return endpoint in self.endpoints 

623 

624 def _should_use_fr_error_handler(self): 

625 """ 

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

627 

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

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

630 method currently handles 404 and 405 errors. 

631 

632 :return: bool 

633 """ 

634 adapter = current_app.create_url_adapter(request) 

635 

636 try: 

637 adapter.match() 

638 except MethodNotAllowed as e: 

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

640 valid_route_method = e.valid_methods[0] 

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

642 return self.owns_endpoint(rule.endpoint) 

643 except NotFound: 

644 return self.catch_all_404s 

645 except Exception: 

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

647 pass 

648 

649 def _has_fr_route(self): 

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

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

652 if self._should_use_fr_error_handler(): 

653 return True 

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

655 if not request.url_rule: 

656 return False 

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

658 

659 def error_router(self, original_handler, e): 

660 """ 

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

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

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

664 app's original error handler will be dispatched. 

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

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

667 back onto the original_handler as last resort. 

668 

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

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

671 """ 

672 if self._has_fr_route(): 

673 try: 

674 return self.handle_error(e) 

675 except Exception as f: 

676 return original_handler(f) 

677 return original_handler(e) 

678 

679 def _propagate_exceptions(self): 

680 """ 

681 Returns the value of the ``PROPAGATE_EXCEPTIONS`` configuration 

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

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

684 we still need it for our error handlers. 

685 """ 

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

687 if rv is not None: 

688 return rv 

689 return current_app.testing or current_app.debug 

690 

691 def handle_error(self, e): 

692 """ 

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

694 with the appropriate HTTP status code and body. 

695 

696 :param Exception e: the raised Exception object 

697 

698 """ 

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

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

701 if ( 

702 not isinstance(e, HTTPException) 

703 and self._propagate_exceptions() 

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

705 ): 

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

707 if exc_value is e: 

708 raise 

709 else: 

710 raise e 

711 

712 include_message_in_response = current_app.config.get( 

713 "ERROR_INCLUDE_MESSAGE", True 

714 ) 

715 default_data = {} 

716 

717 headers = Headers() 

718 

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

720 if isinstance(e, typecheck): 

721 result = handler(e) 

722 default_data, code, headers = unpack( 

723 result, HTTPStatus.INTERNAL_SERVER_ERROR 

724 ) 

725 break 

726 else: 

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

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

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

730 

731 if isinstance(e, HTTPException): 

732 code = None 

733 if e.code is not None: 

734 code = HTTPStatus(e.code) 

735 elif e.response is not None: 

736 code = HTTPStatus(e.response.status_code) 

737 if include_message_in_response: 

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

739 headers = e.get_response().headers 

740 elif self._default_error_handler: 

741 result = self._default_error_handler(e) 

742 default_data, code, headers = unpack( 

743 result, HTTPStatus.INTERNAL_SERVER_ERROR 

744 ) 

745 else: 

746 code = HTTPStatus.INTERNAL_SERVER_ERROR 

747 if include_message_in_response: 

748 default_data = { 

749 "message": code.phrase, 

750 } 

751 

752 if include_message_in_response: 

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

754 

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

756 fallback_mediatype = None 

757 

758 if code >= HTTPStatus.INTERNAL_SERVER_ERROR: 

759 exc_info = sys.exc_info() 

760 if exc_info[1] is None: 

761 exc_info = None 

762 current_app.log_exception(exc_info) 

763 

764 elif ( 

765 code == HTTPStatus.NOT_FOUND 

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

767 and include_message_in_response 

768 ): 

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

770 

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

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

773 # make_response uses a representation we support as the 

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

775 # another NotAcceptable error). 

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

777 fallback_mediatype = ( 

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

779 ) 

780 

781 # Remove blacklisted headers 

782 for header in HEADERS_BLACKLIST: 

783 headers.pop(header, None) 

784 

785 resp = self.make_response( 

786 data, code, headers, fallback_mediatype=fallback_mediatype 

787 ) 

788 

789 if code == HTTPStatus.UNAUTHORIZED: 

790 resp = self.unauthorized(resp) 

791 return resp 

792 

793 def _help_on_404(self, message=None): 

794 rules = dict( 

795 [ 

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

797 for rule in current_app.url_map.iter_rules() 

798 ] 

799 ) 

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

801 if close_matches: 

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

803 message = "".join( 

804 ( 

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

806 "You have requested this URI [", 

807 request.path, 

808 "] but did you mean ", 

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

810 " ?", 

811 ) 

812 ) 

813 return message 

814 

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

816 """ 

817 Serialize the API as Postman collection (v1) 

818 

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

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

821 

822 """ 

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

824 

825 @property 

826 def payload(self): 

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

828 return request.get_json() 

829 

830 @property 

831 def refresolver(self): 

832 if not self._refresolver: 

833 # Create a registry that can resolve references within our schema 

834 registry = Registry() 

835 schema = self.__schema__ 

836 

837 # If schema has definitions, register it 

838 if "definitions" in schema: 

839 schema_id = schema.get("$id", "http://localhost/schema.json") 

840 registry = registry.with_resource(schema_id, schema) 

841 else: 

842 # If no definitions in schema, register all models individually 

843 for name, model in self.models.items(): 

844 model_schema = model.__schema__ 

845 # Add $id to the model schema so it can be referenced 

846 if "$id" not in model_schema: 

847 model_schema = model_schema.copy() 

848 model_schema["$id"] = ( 

849 f"http://localhost/schema.json#/definitions/{name}" 

850 ) 

851 registry = registry.with_resource( 

852 f"http://localhost/schema.json#/definitions/{name}", 

853 model_schema, 

854 ) 

855 

856 # Also register the root schema with definitions 

857 if self.models: 

858 definitions = {} 

859 for name, model in self.models.items(): 

860 definitions[name] = model.__schema__ 

861 

862 schema_with_definitions = { 

863 "$id": "http://localhost/schema.json", 

864 "definitions": definitions, 

865 } 

866 registry = registry.with_resource( 

867 "http://localhost/schema.json", schema_with_definitions 

868 ) 

869 

870 self._refresolver = registry 

871 return self._refresolver 

872 

873 @staticmethod 

874 def _blueprint_setup_add_url_rule_patch( 

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