Add jsonschema support

Author: Ron Serruya

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = aws-glue-schema-registry
-version = 1.0.0
+version = 1.1.0
 description = Use the AWS Glue Schema Registry.
 long_description = file: README.md
 long_description_content_type = text/markdown
@@ -34,6 +34,8 @@ install_requires =
     boto3>=1.18.48
     typing-extensions>=3.10.0.2;python_version<"3.8"
     fastavro>=1.4.5
+    orjson~=3.6.0
+    fastjsonschema~=2.15
 setup_requires =
     setuptools
 
@@ -52,3 +54,6 @@ ignore_missing_imports = True
 
 [mypy-boto3.*]
 ignore_missing_imports = True
+
+[mypy-fastjsonschema.*]
+ignore_missing_imports = True

src/aws_schema_registry/jsonschema.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Union
+
+import orjson
+
+import fastjsonschema
+
+from aws_schema_registry.schema import DataFormat, Schema, ValidationError
+
+
+class JsonSchema(Schema):
+    """Implementation of the `Schema` protocol for JSON schemas.
+
+    Arguments:
+        definition: the schema, either as a parsed dict or a string
+    """
+
+    def __init__(self, definition: Union[str, dict]):
+        if isinstance(definition, str):
+            self._dict = orjson.loads(definition)
+        else:
+            self._dict = definition
+        self._compiled_validation_method = fastjsonschema.compile(self._dict)
+
+    def __hash__(self):
+        return hash(str(self))
+
+    def __eq__(self, other):
+        return isinstance(other, JsonSchema) and \
+               self._dict == other._dict
+
+    def __str__(self):
+        return orjson.dumps(self._dict).decode()
+
+    def __repr__(self):
+        return '<JsonSchema %s>' % self._dict
+
+    @property
+    def data_format(self) -> DataFormat:
+        return 'JSON'
+
+    @property
+    def fqn(self) -> str:
+        return ""
+
+    def read(self, bytes_: bytes):
+        return orjson.loads(bytes_)
+
+    def write(self, data) -> bytes:
+        return orjson.dumps(data)
+
+    def validate(self, data):
+        try:
+            self._compiled_validation_method(data)
+        except fastjsonschema.exceptions.JsonSchemaValueException as e:
+            raise ValidationError(str(e)) from e

src/aws_schema_registry/serde.py

@@ -6,6 +6,7 @@
 from uuid import UUID
 
 from aws_schema_registry.avro import AvroSchema
+from aws_schema_registry.jsonschema import JsonSchema
 from aws_schema_registry.client import SchemaRegistryClient
 from aws_schema_registry.codec import decode, encode, UnknownEncodingException
 from aws_schema_registry.exception import SchemaRegistryException
@@ -61,7 +62,7 @@ def serialize(self, topic: str, data_and_schema: DataAndSchema):
         if data_and_schema is None:
             return None
         if not isinstance(data_and_schema, tuple):
-            raise TypeError('AvroSerializer can only serialize',
+            raise TypeError('KafkaSerializer can only serialize',
                             f' {tuple}, got {type(data_and_schema)}')
         data, schema = data_and_schema
         schema_version = self._get_schema_version(topic, schema)
@@ -136,4 +137,4 @@ def _schema_for_version(self, version: SchemaVersion) -> Schema:
         if version.data_format == 'AVRO':
             return AvroSchema(version.definition)
         elif version.data_format == 'JSON':
-            raise NotImplementedError('JSON schema not supported')
+            return JsonSchema(version.definition)