Source code for webargs.flaskparser

"""Flask request argument parsing module.

Example: ::

    from flask import Flask

    from webargs import fields
    from webargs.flaskparser import use_args

    app = Flask(__name__)

    user_detail_args = {
        'per_page': fields.Int()

    def user_detail(args, uid):
        return ("The user page for user {uid}, showing {per_page} posts.").format(
            uid=uid, per_page=args["per_page"]
from __future__ import annotations
from typing import NoReturn

import flask
from werkzeug.exceptions import HTTPException

import marshmallow as ma

from webargs import core

[docs]def abort(http_status_code, exc=None, **kwargs) -> NoReturn: """Raise a HTTPException for the given http_status_code. Attach any keyword arguments to the exception for later processing. From Flask-Restful. See NOTICE file for license information. """ try: flask.abort(http_status_code) except HTTPException as err: = kwargs # type: ignore err.exc = exc # type: ignore raise err
def is_json_request(req): return core.is_json(req.mimetype)
[docs]class FlaskParser(core.Parser): """Flask request argument parser.""" DEFAULT_UNKNOWN_BY_LOCATION: dict[str, str | None] = { "view_args": ma.RAISE, "path": ma.RAISE, **core.Parser.DEFAULT_UNKNOWN_BY_LOCATION, } __location_map__ = dict( view_args="load_view_args", path="load_view_args", **core.Parser.__location_map__, ) def _raw_load_json(self, req): """Return a json payload from the request for the core parser's load_json Checks the input mimetype and may return 'missing' if the mimetype is non-json, even if the request body is parseable as json.""" if not is_json_request(req): return core.missing return core.parse_json(req.get_data(cache=True)) def _handle_invalid_json_error(self, error, req, *args, **kwargs): abort(400, exc=error, messages={"json": ["Invalid JSON body."]})
[docs] def load_view_args(self, req, schema): """Return the request's ``view_args`` or ``missing`` if there are none.""" return req.view_args or core.missing
[docs] def load_querystring(self, req, schema): """Return query params from the request as a MultiDictProxy.""" return self._makeproxy(req.args, schema)
[docs] def load_form(self, req, schema): """Return form values from the request as a MultiDictProxy.""" return self._makeproxy(req.form, schema)
[docs] def load_headers(self, req, schema): """Return headers from the request as a MultiDictProxy.""" return self._makeproxy(req.headers, schema)
[docs] def load_cookies(self, req, schema): """Return cookies from the request.""" return req.cookies
[docs] def load_files(self, req, schema): """Return files from the request as a MultiDictProxy.""" return self._makeproxy(req.files, schema)
[docs] def handle_error(self, error, req, schema, *, error_status_code, error_headers): """Handles errors during parsing. Aborts the current HTTP request and responds with a 422 error. """ status_code = error_status_code or self.DEFAULT_VALIDATION_STATUS abort( status_code, exc=error, messages=error.messages, schema=schema, headers=error_headers, )
[docs] def get_default_request(self): """Override to use Flask's thread-local request object by default""" return flask.request
parser = FlaskParser() use_args = parser.use_args use_kwargs = parser.use_kwargs