Source code for falcon.media.validators.jsonschema

from __future__ import annotations

from functools import wraps
from inspect import iscoroutinefunction
from typing import Any, Callable, Dict, Optional, TYPE_CHECKING

import falcon

try:
    import jsonschema
except ImportError:  # pragma: nocover
    pass

if TYPE_CHECKING:
    import falcon as wsgi
    from falcon import asgi

Schema = Optional[Dict[str, Any]]
ResponderMethod = Callable[..., Any]


[docs] def validate( req_schema: Schema = None, resp_schema: Schema = None ) -> Callable[[ResponderMethod], ResponderMethod]: """Validate ``req.media`` using JSON Schema. This decorator provides standard JSON Schema validation via the ``jsonschema`` package available from PyPI. Semantic validation via the *format* keyword is enabled for the default checkers implemented by ``jsonschema.FormatChecker``. In the case of failed request media validation, an instance of :class:`~falcon.MediaValidationError` is raised by the decorator. By default, this error is rendered as a 400 (:class:`~falcon.HTTPBadRequest`) response with the ``title`` and ``description`` attributes explaining the validation failure, but this behavior can be modified by adding a custom error :func:`handler <falcon.App.add_error_handler>` for :class:`~falcon.MediaValidationError`. Note: The ``jsonschema`` package must be installed separately in order to use this decorator, as Falcon does not install it by default. See `json-schema.org <http://json-schema.org/>`_ for more information on defining a compatible dictionary. Keyword Args: req_schema (dict): A dictionary that follows the JSON Schema specification. The request will be validated against this schema. resp_schema (dict): A dictionary that follows the JSON Schema specification. The response will be validated against this schema. Example: .. tab-set:: .. tab-item:: WSGI .. code:: python from falcon.media.validators import jsonschema # -- snip -- @jsonschema.validate(my_post_schema) def on_post(self, req, resp): # -- snip -- .. tab-item:: ASGI .. code:: python from falcon.media.validators import jsonschema # -- snip -- @jsonschema.validate(my_post_schema) async def on_post(self, req, resp): # -- snip -- """ def decorator(func: ResponderMethod) -> ResponderMethod: if iscoroutinefunction(func): return _validate_async(func, req_schema, resp_schema) return _validate(func, req_schema, resp_schema) return decorator
def _validate( func: ResponderMethod, req_schema: Schema = None, resp_schema: Schema = None ) -> ResponderMethod: @wraps(func) def wrapper( self: Any, req: wsgi.Request, resp: wsgi.Response, *args: Any, **kwargs: Any ) -> Any: if req_schema is not None: try: jsonschema.validate( req.media, req_schema, format_checker=jsonschema.FormatChecker() ) except jsonschema.ValidationError as ex: raise falcon.MediaValidationError( title='Request data failed validation', description=ex.message ) from ex result = func(self, req, resp, *args, **kwargs) if resp_schema is not None: try: jsonschema.validate( resp.media, resp_schema, format_checker=jsonschema.FormatChecker() ) except jsonschema.ValidationError as ex: raise falcon.HTTPInternalServerError( title='Response data failed validation' # Do not return 'e.message' in the response to # prevent info about possible internal response # formatting bugs from leaking out to users. ) from ex return result return wrapper def _validate_async( func: ResponderMethod, req_schema: Schema = None, resp_schema: Schema = None ) -> ResponderMethod: @wraps(func) async def wrapper( self: Any, req: asgi.Request, resp: asgi.Response, *args: Any, **kwargs: Any ) -> Any: if req_schema is not None: m = await req.get_media() try: jsonschema.validate( m, req_schema, format_checker=jsonschema.FormatChecker() ) except jsonschema.ValidationError as ex: raise falcon.MediaValidationError( title='Request data failed validation', description=ex.message ) from ex result = await func(self, req, resp, *args, **kwargs) if resp_schema is not None: try: jsonschema.validate( resp.media, resp_schema, format_checker=jsonschema.FormatChecker() ) except jsonschema.ValidationError as ex: raise falcon.HTTPInternalServerError( title='Response data failed validation' # Do not return 'e.message' in the response to # prevent info about possible internal response # formatting bugs from leaking out to users. ) from ex return result return wrapper