Quart webargs parser

quart_parser.py

# -*- coding: utf-8 -*-
"""Quart request argument parsing module.

Example: ::

    from quart import Quart

    from webargs import fields
    from webargs.quartparser import use_args

    app = Quart(__name__)

    hello_args = {
        'name': fields.Str(required=True)
    }

    @app.route('/')
    @use_args(hello_args)
    def index(args):
        return 'Hello ' + args['name']

    app.run()
"""
import typing

import quart
from marshmallow.schema import Schema
from quart.exceptions import HTTPException
from webargs import core
from webargs.asyncparser import AsyncParser
from webargs.core import json
from webargs.multidictproxy import MultiDictProxy


def abort(http_status_code, exc=None, **kwargs):
    """Raise a HTTPException for the given http_status_code. Attach any keyword
    arguments to the exception for later processing.
    """
    try:
        quart.abort(http_status_code)
    except HTTPException as err:
        err.data = kwargs
        err.exc = exc
        raise err


def is_json_request(req):
    return core.is_json(req.mimetype)


class QuartParser(AsyncParser):
    """Quart request argument parser."""

    __location_map__ = dict(
        view_args="parse_view_args",
        **core.Parser.__location_map__
    )

    def load_querystring(self, req, schema: Schema) -> MultiDictProxy:
        """Return query params from the request as a MultiDictProxy."""
        return MultiDictProxy(req.query, schema)

    async def load_form(self, req, schema: Schema) -> MultiDictProxy:
        """Return form values from the request as a MultiDictProxy."""
        post_data = await req.form
        return MultiDictProxy(post_data, schema)

    async def load_json_or_form(
        self, req, schema: Schema
    ) -> typing.Union[typing.Dict, MultiDictProxy]:
        data = await self.load_json(req, schema)
        if data is not core.missing:
            return data
        return await self.load_form(req, schema)

    async def load_json(self, req, schema: Schema):
        """Return a parsed json payload from the request."""
        body = await req.body
        if not (body and is_json_request(req)):
            return core.missing
        try:
            return await req.get_json()
        except json.JSONDecodeError as e:
            if e.doc == "":
                return core.missing
            else:
                return self.handle_invalid_json_error(e, req)
        except UnicodeDecodeError as e:
            return self._handle_invalid_json_error(e, req)

    def load_headers(self, req, schema: Schema) -> MultiDictProxy:
        """Return headers from the request as a MultiDictProxy."""
        return MultiDictProxy(req.headers, schema)

    def load_cookies(self, req, schema: Schema) -> MultiDictProxy:
        """Return cookies from the request as a MultiDictProxy."""
        return MultiDictProxy(req.cookies, schema)

    async def load_files(self, req, schema: Schema) -> MultiDictProxy:
        """Return files from the request as a MultiDictProxy."""
        return MultiDictProxy(await req.files, schema)

    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,
        )

    def handle_invalid_json_error(self, error, req, *args, **kwargs):
        abort(400, exc=error, messages={"json": ["Invalid JSON body."]})

    def get_default_request(self):
        """Override to use Quart's thread-local request object by default"""
        return quart.request


parser = QuartParser()
use_args = parser.use_args
use_kwargs = parser.use_kwargs