Source code for webargs.falconparser

"""Falcon request argument parsing module."""

import falcon
import marshmallow as ma
from falcon.util.uri import parse_query_string

from webargs import core

HTTP_422 = "422 Unprocessable Entity"

# Mapping of int status codes to string status
status_map = {422: HTTP_422}

# Collect all exceptions from falcon.status_codes
def _find_exceptions():
    for name in filter(lambda n: n.startswith("HTTP"), dir(falcon.status_codes)):
        status = getattr(falcon.status_codes, name)
        status_code = int(status.split(" ")[0])
        status_map[status_code] = status

del _find_exceptions

def is_json_request(req: falcon.Request):
    content_type = req.get_header("Content-Type")
    return content_type and core.is_json(content_type)

# NOTE: Adapted from falcon.request.Request._parse_form_urlencoded
def parse_form_body(req: falcon.Request):
    if (
        req.content_type is not None
        and "application/x-www-form-urlencoded" in req.content_type
        body = or 0)
            body = body.decode("ascii")
        except UnicodeDecodeError:
            body = None
                "Non-ASCII characters found in form body "
                "with Content-Type of "
                "application/x-www-form-urlencoded. Body "
                "will be ignored."

        if body:
            return parse_query_string(body, keep_blank=req.options.keep_blank_qs_values)

    return core.missing

[docs] class HTTPError(falcon.HTTPError): """HTTPError that stores a dictionary of validation error messages.""" def __init__(self, status, errors, *args, **kwargs): self.errors = errors super().__init__(status, *args, **kwargs)
[docs] def to_dict(self, *args, **kwargs): """Override `falcon.HTTPError` to include error messages in responses.""" ret = super().to_dict(*args, **kwargs) if self.errors is not None: ret["errors"] = self.errors return ret
[docs] class FalconParser(core.Parser[falcon.Request]): """Falcon request argument parser. Defaults to using the `media` location. See :py:meth:`~FalconParser.load_media` for details on the media location.""" # by default, Falcon will use the 'media' location to load data # # this effectively looks the same as loading JSON data by default, but if # you add a handler for a different media type to Falcon, webargs will # automatically pick up on that capability DEFAULT_LOCATION = "media" DEFAULT_UNKNOWN_BY_LOCATION = dict( media=ma.RAISE, **core.Parser.DEFAULT_UNKNOWN_BY_LOCATION ) __location_map__ = dict(media="load_media", **core.Parser.__location_map__) # Note on the use of MultiDictProxy throughout: # Falcon parses query strings and form values into ordinary dicts, but with # the values listified where appropriate # it is still therefore necessary in these cases to wrap them in # MultiDictProxy because we need to use the schema to determine when single # values should be wrapped in lists due to the type of the destination # field
[docs] def load_querystring(self, req: falcon.Request, schema): """Return query params from the request as a MultiDictProxy.""" return self._makeproxy(req.params, schema)
[docs] def load_form(self, req: falcon.Request, schema): """Return form values from the request as a MultiDictProxy .. note:: The request stream will be read and left at EOF. """ form = parse_form_body(req) if form is core.missing: return form return self._makeproxy(form, schema)
[docs] def load_media(self, req: falcon.Request, schema): """Return data unpacked and parsed by one of Falcon's media handlers. By default, Falcon only handles JSON payloads. To configure additional media handlers, see the `Falcon documentation on media types`__. .. _FalconMedia: __ FalconMedia_ .. note:: The request stream will be read and left at EOF. """ # if there is no body, return missing instead of erroring if req.content_length in (None, 0): return core.missing return
def _raw_load_json(self, req: falcon.Request): """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) or req.content_length in (None, 0): return core.missing body = if body: return core.parse_json(body) return core.missing
[docs] def load_headers(self, req: falcon.Request, schema): """Return headers from the request.""" # Falcon only exposes headers as a dict (not multidict) return req.headers
[docs] def load_cookies(self, req: falcon.Request, schema): """Return cookies from the request.""" # Cookies are expressed in Falcon as a dict, but the possibility of # multiple values for a cookie is preserved internally -- if desired in # the future, webargs could add a MultiDict type for Cookies here built # from (req, schema), but Falcon does not provide one out of the box return req.cookies
[docs] def get_request_from_view_args(self, view, args, kwargs): """Get request from a resource method's arguments. Assumes that request is the second argument. """ req = args[1] if not isinstance(req, falcon.Request): raise TypeError("Argument is not a falcon.Request") return req
[docs] def load_files(self, req: falcon.Request, schema): raise NotImplementedError( f"Parsing files not yet supported by {self.__class__.__name__}" )
[docs] def handle_error( self, error, req: falcon.Request, schema, *, error_status_code, error_headers ): """Handles errors during parsing.""" status = status_map.get(error_status_code or self.DEFAULT_VALIDATION_STATUS) if status is None: raise LookupError(f"Status code {error_status_code} not supported") raise HTTPError(status, errors=error.messages, headers=error_headers)
def _handle_invalid_json_error(self, error, req: falcon.Request, *args, **kwargs): status = status_map[400] messages = {"json": ["Invalid JSON body."]} raise HTTPError(status, errors=messages)
parser = FalconParser() use_args = parser.use_args use_kwargs = parser.use_kwargs