feat: generalize validators and support jsonschema-rs

falcon/media/validators/__init__.py

@@ -1 +1,2 @@
 from . import jsonschema
+from . import jsonschema_rs

falcon/media/validators/base.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+from functools import wraps
+from inspect import iscoroutinefunction
+from abc import abstractmethod
+from typing import Any, Tuple, Type, Union, Optional
+
+import falcon
+
+
+class Validator:
+ """Base validator class."""
+
+ exceptions: Union[Tuple[Type[Exception], ...], Type[Exception]]
+ """The exceptions raised by the validation library"""
+
+ @classmethod
+ @abstractmethod
+ def from_schema(cls, schema: Any) -> Validator:
+ """Construct the class from a schema object."""
+
+ @abstractmethod
+ def validate(self, media: Any) -> None:
+ """Validates the input media"""
+
+ @abstractmethod
+ def get_exception_message(self, exception: Exception) -> Optional[str]:
+ """Returns a message from an exception"""
+
+
+def validator_factory(
+ validator: Type[Validator], req_schema: Any, resp_schema: Any, is_async: bool
+):
+ def decorator(func):
+ if iscoroutinefunction(func) or is_async:
+ return _validate_async(validator, func, req_schema, resp_schema)
+
+ return _validate(validator, func, req_schema, resp_schema)
+
+ return decorator
+
+
+def _validate(validator: Type[Validator], func, req_schema: Any, resp_schema: Any):
+ req_validator = None if req_schema is None else validator.from_schema(req_schema)
+ resp_validator = None if resp_schema is None else validator.from_schema(resp_schema)
+
+ @wraps(func)
+ def wrapper(self, req, resp, *args, **kwargs):
+ if req_validator is not None:
+ try:
+ req_validator.validate(req.media)
+ except req_validator.exceptions as ex:
+ raise falcon.MediaValidationError(
+ title='Request data failed validation',
+ description=req_validator.get_exception_message(ex),
+ ) from ex
+
+ result = func(self, req, resp, *args, **kwargs)
+
+ if resp_validator is not None:
+ try:
+ resp_validator.validate(resp.media)
+ except resp_validator.exceptions 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(
+ validator: Type[Validator], func, req_schema: Any, resp_schema: Any
+):
+ req_validator = None if req_schema is None else validator.from_schema(req_schema)
+ resp_validator = None if resp_schema is None else validator.from_schema(resp_schema)
+
+ @wraps(func)
+ async def wrapper(self, req, resp, *args, **kwargs):
+ if req_validator is not None:
+ m = await req.get_media()
+
+ try:
+ req_validator.validate(m)
+ except req_validator.exceptions as ex:
+ raise falcon.MediaValidationError(
+ title='Request data failed validation',
+ description=req_validator.get_exception_message(ex),
+ ) from ex
+
+ result = await func(self, req, resp, *args, **kwargs)
+
+ if resp_validator is not None:
+ try:
+ resp_validator.validate(resp.media)
+ except resp_validator.exceptions 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

falcon/media/validators/jsonschema.py

@@ -1,15 +1,16 @@
-from functools import wraps
-from inspect import iscoroutinefunction
+from __future__ import annotations
 
-import falcon
+from typing import Any
+
+from . import base as _base
 
 try:
  import jsonschema
 except ImportError: # pragma: nocover
  pass
 
 
-def validate(req_schema=None, resp_schema=None, is_async=False):
+def validate(req_schema: Any = None, resp_schema: Any = None, is_async: bool = False):
  """Validate ``req.media`` using JSON Schema.
 
  This decorator provides standard JSON Schema validation via the
@@ -99,78 +100,24 @@ async def on_post(self, req, resp):
 
  """
 
- def decorator(func):
- if iscoroutinefunction(func) or is_async:
- return _validate_async(func, req_schema, resp_schema)
-
- return _validate(func, req_schema, resp_schema)
-
- return decorator
-
-
-def _validate(func, req_schema=None, resp_schema=None):
- @wraps(func)
- def wrapper(self, req, resp, *args, **kwargs):
- 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, req_schema=None, resp_schema=None):
- @wraps(func)
- async def wrapper(self, req, resp, *args, **kwargs):
- 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
+ return _base.validator_factory(
+ JsonSchemaValidator, req_schema, resp_schema, is_async
+ )
+
+
+class JsonSchemaValidator(_base.Validator):
+ def __init__(self, schema: Any) -> None:
+ self.schema = schema
+ self.exceptions = jsonschema.ValidationError
+
+ @classmethod
+ def from_schema(cls, schema: Any) -> JsonSchemaValidator:
+ return cls(schema)
+
+ def validate(self, media: Any) -> None:
+ jsonschema.validate(
+ media, self.schema, format_checker=jsonschema.FormatChecker()
+ )
+
+ def get_exception_message(self, exception: jsonschema.ValidationError):
+ return exception.message

falcon/media/validators/jsonschema_rs.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+from typing import Any
+
+from . import base as _base
+
+try:
+ import jsonschema_rs
+except ImportError: # pragma: nocover
+ pass
+
+
+def validate(req_schema: Any = None, resp_schema: Any = None, is_async: bool = False):
+ """Validate ``req.media`` using JSON Schema.
+
+ This decorator provides standard JSON Schema validation via the
+ ``jsonschema_rs`` package available from PyPI.
+
+ 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_rs`` package must be installed separately in order to use
+ this decorator, as Falcon does not install it by default.
+
+ See `jsonschema_rs PyPi <https://pypi.org/project/jsonschema-rs/>`_ for more
+ information on defining a compatible dictionary.
+
+ Keyword Args:
+ req_schema (dict or str): A dictionary that follows the JSON
+ Schema specification. The request will be validated against this
+ schema.
+ Can be also a json string that will be loaded by the jsonschema_rs library
+ resp_schema (dict or str): A dictionary that follows the JSON
+ Schema specification. The response will be validated against this
+ schema.
+ Can be also a json string that will be loaded by the jsonschema_rs library
+ is_async (bool): Set to ``True`` for ASGI apps to provide a hint that
+ the decorated responder is a coroutine function (i.e., that it
+ is defined with ``async def``) or that it returns an awaitable
+ coroutine object.
+
+ Normally, when the function source is declared using ``async def``,
+ the resulting function object is flagged to indicate it returns a
+ coroutine when invoked, and this can be automatically detected.
+ However, it is possible to use a regular function to return an
+ awaitable coroutine object, in which case a hint is required to let
+ the framework know what to expect. Also, a hint is always required
+ when using a cythonized coroutine function, since Cython does not
+ flag them in a way that can be detected in advance, even when the
+ function is declared using ``async def``.
+
+ Example:
+
+ .. tabs::
+
+ .. tab:: WSGI
+
+ .. code:: python
+
+ from falcon.media.validators import jsonschema_rs
+
+ # -- snip --
+
+ @jsonschema_rs.validate(my_post_schema)
+ def on_post(self, req, resp):
+
+ # -- snip --
+
+ .. tab:: ASGI
+
+ .. code:: python
+
+ from falcon.media.validators import jsonschema_rs
+
+ # -- snip --
+
+ @jsonschema_rs.validate(my_post_schema)
+ async def on_post(self, req, resp):
+
+ # -- snip --
+
+ .. tab:: ASGI (Cythonized App)
+
+ .. code:: python
+
+ from falcon.media.validators import jsonschema_rs
+
+ # -- snip --
+
+ @jsonschema_rs.validate(my_post_schema, is_async=True)
+ async def on_post(self, req, resp):
+
+ # -- snip --
+
+ """
+
+ return _base.validator_factory(
+ JsonSchemaRsValidator, req_schema, resp_schema, is_async
+ )
+
+
+class JsonSchemaRsValidator(_base.Validator):
+ def __init__(self, schema: Any) -> None:
+ self.schema = schema
+ if isinstance(schema, str):
+ self.validator = jsonschema_rs.JSONSchema.from_str(schema)
+ else:
+ self.validator = jsonschema_rs.JSONSchema(schema)
+ self.exceptions = jsonschema_rs.ValidationError
+
+ @classmethod
+ def from_schema(cls, schema: Any) -> JsonSchemaRsValidator:
+ return cls(schema)
+
+ def validate(self, media: Any) -> None:
+ self.validator.validate(media)
+
+ def get_exception_message(self, exception: jsonschema_rs.ValidationError):
+ return exception.message