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