Support jsonschema validation in json serdes (#414)

Author: ovv

LICENSES/LICENSE.jsonschema

@@ -0,0 +1,19 @@
+Copyright (c) 2013 Julian Berman
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

docs/advanced/serialization.md

@@ -48,3 +48,27 @@ output_topic = app.topic('output', value_serializer=JSONSerializer())
 ```
 
 You can find all available serializers in `quixstreams.models.serializers` module.
+
+## Jsonschema support
+
+The json serializer and deserializer support validation of the data against a jsonschema.
+
+```python
+from jsonschema import Draft202012Validator
+
+from quixstreams import Application
+from quixstreams.models import JSONDeserializer, JSONSerializer
+
+MY_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "name": {"type": "string"},
+        "id": {"type": "number"},
+    },
+    "required": ["id"],
+}
+
+app = Application(broker_address='localhost:9092', consumer_group='consumer')
+input_topic = app.topic('input', value_deserializer=JSONDeserializer(schema=MY_SCHEMA))
+output_topic = app.topic('output', value_serializer=JSONSerializer(schema=MY_SCHEMA))
+```

quixstreams/models/serializers/json.py

@@ -1,5 +1,7 @@
 from typing import Callable, Union, Mapping, Optional, Any, Iterable
 
+from jsonschema import ValidationError, SchemaError, Validator, Draft202012Validator
+
 from quixstreams.utils.json import (
     dumps as default_dumps,
     loads as default_loads,
@@ -14,18 +16,39 @@ class JSONSerializer(Serializer):
     def __init__(
         self,
         dumps: Callable[[Any], Union[str, bytes]] = default_dumps,
+        schema: Optional[Mapping] = None,
+        validator: Optional[Validator] = None,
     ):
         """
         Serializer that returns data in json format.
         :param dumps: a function to serialize objects to json.
             Default - :py:func:`quixstreams.utils.json.dumps`
+        :param schema: A schema used to validate the data using [`jsonschema.Draft202012Validator`](https://python-jsonschema.readthedocs.io/en/stable/api/jsonschema/validators/#jsonschema.validators.Draft202012Validator).
+            Default - `None`
+        :param validator: A jsonschema validator used to validate the data. Takes precedences over the schema.
+            Default - `None`
         """
+
+        if schema and not validator:
+            validator = Draft202012Validator(schema)
+
+        super().__init__()
         self._dumps = dumps
+        self._validator = validator
+
+        if self._validator:
+            self._validator.check_schema(self._validator.schema)
 
     def __call__(self, value: Any, ctx: SerializationContext) -> Union[str, bytes]:
         return self._to_json(value)
 
     def _to_json(self, value: Any):
+        if self._validator:
+            try:
+                self._validator.validate(value)
+            except ValidationError as exc:
+                raise SerializationError(str(exc)) from exc
+
         try:
             return self._dumps(value)
         except (ValueError, TypeError) as exc:
@@ -36,20 +59,42 @@ class JSONDeserializer(Deserializer):
     def __init__(
         self,
         loads: Callable[[Union[bytes, bytearray]], Any] = default_loads,
+        schema: Optional[Mapping] = None,
+        validator: Optional[Validator] = None,
     ):
         """
         Deserializer that parses data from JSON
 
         :param loads: function to parse json from bytes.
             Default - :py:func:`quixstreams.utils.json.loads`.
+        :param schema: A schema used to validate the data using [`jsonschema.Draft202012Validator`](https://python-jsonschema.readthedocs.io/en/stable/api/jsonschema/validators/#jsonschema.validators.Draft202012Validator).
+            Default - `None`
+        :param validator: A jsonschema validator used to validate the data. Takes precedences over the schema.
+            Default - `None`
         """
+
+        if schema and not validator:
+            validator = Draft202012Validator(schema)
+
         super().__init__()
         self._loads = loads
+        self._validator = validator
+
+        if self._validator:
+            self._validator.check_schema(self._validator.schema)
 
     def __call__(
         self, value: bytes, ctx: SerializationContext
     ) -> Union[Iterable[Mapping], Mapping]:
         try:
-            return self._loads(value)
+            data = self._loads(value)
         except (ValueError, TypeError) as exc:
             raise SerializationError(str(exc)) from exc
+
+        if self._validator:
+            try:
+                self._validator.validate(data)
+            except ValidationError as exc:
+                raise SerializationError(str(exc)) from exc
+
+        return data

requirements.txt

@@ -5,3 +5,4 @@ typing_extensions>=4.8
 orjson>=3.9,<4
 pydantic>=2.7,<2.8
 pydantic-settings>=2.3,<2.4
+jsonschema>=4.0

tests/test_quixstreams/test_models/test_serializers.py

@@ -1,5 +1,7 @@
 import pytest
 
+import jsonschema
+
 from quixstreams.models import (
     IntegerSerializer,
     SerializationContext,
@@ -20,6 +22,15 @@
 
 dummy_context = SerializationContext(topic="topic")
 
+JSONSCHEMA_TEST_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "name": {"type": "string"},
+        "id": {"type": "number"},
+    },
+    "required": ["name"],
+}
+
 
 class TestSerializers:
     @pytest.mark.parametrize(
@@ -34,6 +45,18 @@ class TestSerializers:
             (BytesSerializer(), b"abc", b"abc"),
             (JSONSerializer(), {"a": 123}, b'{"a":123}'),
             (JSONSerializer(), [1, 2, 3], b"[1,2,3]"),
+            (
+                JSONSerializer(schema=JSONSCHEMA_TEST_SCHEMA),
+                {"id": 10, "name": "foo"},
+                b'{"id":10,"name":"foo"}',
+            ),
+            (
+                JSONSerializer(
+                    validator=jsonschema.Draft202012Validator(JSONSCHEMA_TEST_SCHEMA)
+                ),
+                {"id": 10, "name": "foo"},
+                b'{"id":10,"name":"foo"}',
+            ),
         ],
     )
     def test_serialize_success(self, serializer: Serializer, value, expected):
@@ -50,12 +73,28 @@ def test_serialize_success(self, serializer: Serializer, value, expected):
             (StringSerializer(), {"a": 123}),
             (JSONSerializer(), object()),
             (JSONSerializer(), complex(1, 2)),
+            (
+                JSONSerializer(schema=JSONSCHEMA_TEST_SCHEMA),
+                {"id": 10},
+            ),
+            (
+                JSONSerializer(
+                    validator=jsonschema.Draft202012Validator(JSONSCHEMA_TEST_SCHEMA)
+                ),
+                {"id": 10},
+            ),
         ],
     )
     def test_serialize_error(self, serializer: Serializer, value):
         with pytest.raises(SerializationError):
             serializer(value, ctx=dummy_context)
 
+    def test_invalid_jsonschema(self):
+        with pytest.raises(jsonschema.SchemaError):
+            JSONSerializer(
+                validator=jsonschema.Draft202012Validator({"type": "invalid"})
+            )
+
 
 class TestDeserializers:
     @pytest.mark.parametrize(
@@ -70,6 +109,18 @@ class TestDeserializers:
             (BytesDeserializer(), b"123123", b"123123"),
             (JSONDeserializer(), b"123123", 123123),
             (JSONDeserializer(), b'{"a":"b"}', {"a": "b"}),
+            (
+                JSONDeserializer(schema=JSONSCHEMA_TEST_SCHEMA),
+                b'{"id":10,"name":"foo"}',
+                {"id": 10, "name": "foo"},
+            ),
+            (
+                JSONDeserializer(
+                    validator=jsonschema.Draft202012Validator(JSONSCHEMA_TEST_SCHEMA)
+                ),
+                b'{"id":10,"name":"foo"}',
+                {"id": 10, "name": "foo"},
+            ),
         ],
     )
     def test_deserialize_no_column_name_success(
@@ -84,8 +135,27 @@ def test_deserialize_no_column_name_success(
             (IntegerDeserializer(), b'{"abc": "abc"}'),
             (DoubleDeserializer(), b"abc"),
             (JSONDeserializer(), b"{"),
+            (
+                JSONDeserializer(schema=JSONSCHEMA_TEST_SCHEMA),
+                b'{"id":10}',
+            ),
+            (
+                JSONDeserializer(
+                    validator=jsonschema.Draft202012Validator(JSONSCHEMA_TEST_SCHEMA)
+                ),
+                b'{"id":10}',
+            ),
         ],
     )
     def test_deserialize_error(self, deserializer: Deserializer, value):
         with pytest.raises(SerializationError):
             deserializer(value, ctx=dummy_context)
+
+    def test_invalid_jsonschema(self):
+        with pytest.raises(jsonschema.SchemaError):
+            JSONDeserializer(
+                validator=jsonschema.Draft202012Validator({"type": "invalid"})
+            )
+
+        with pytest.raises(jsonschema.SchemaError):
+            JSONDeserializer(schema={"type": "invalid"})