Add validation to schema interface

Author: cdebost

src/aws_schema_registry/__init__.py

@@ -1,6 +1,8 @@
 from .client import SchemaRegistryClient
 from .exception import SchemaRegistryException
-from .schema import CompatibilityMode, DataFormat, Schema, SchemaVersion
+from .schema import (
+    CompatibilityMode, DataFormat, Schema, SchemaVersion, ValidationError
+)
 from .serde import (
     DataAndSchema, SchemaRegistryDeserializer, SchemaRegistrySerializer
 )
@@ -16,5 +18,6 @@
     'SchemaRegistryDeserializer',
     'SchemaRegistryException',
     'SchemaRegistrySerializer',
-    'SchemaVersion'
+    'SchemaVersion',
+    'ValidationError'
 ]

src/aws_schema_registry/avro.py

@@ -5,7 +5,7 @@
 
 import fastavro
 
-from aws_schema_registry.schema import DataFormat, Schema
+from aws_schema_registry.schema import DataFormat, Schema, ValidationError
 
 
 class AvroSchema(Schema):
@@ -63,3 +63,12 @@ def write(self, data) -> bytes:
         value = b.getvalue()
         b.close()
         return value
+
+    def validate(self, data):
+        try:
+            fastavro.validate(data, self._parsed)
+        except Exception as e:
+            # the message will contain space characters, json.loads + str is a
+            # (relatively inefficient) way to remove them
+            detail: list[str] = json.loads(str(e))
+            raise ValidationError(str(detail)) from e

src/aws_schema_registry/schema.py

@@ -55,6 +55,18 @@ def read(self, bytes_: bytes) -> Any:
     def write(self, data) -> bytes:
         """Write a record into bytes."""
 
+    def validate(self, data) -> None:
+        """Raise a ValidationException if the data is invalid."""
+
+
+class ValidationError(Exception):
+    """Raised when a schema's `validate` is called on invalid data.
+
+    The error need not contain *every* validation error, just the first that
+    classifies the data as invalid.
+    """
+    pass
+
 
 @dataclass
 class SchemaVersion:

tests/test_avro.py

@@ -1,3 +1,6 @@
+import pytest
+
+from aws_schema_registry import ValidationError
 from aws_schema_registry.avro import AvroSchema
 
 
@@ -27,3 +30,24 @@ def test_readwrite():
         'age': 900
     }
     assert s.read(s.write(d)) == d
+
+
+def test_validation():
+    s = AvroSchema('''
+{
+  "type": "record",
+  "name": "JediMaster",
+  "fields": [
+    {"name": "name", "type": "string" },
+    {"name": "age", "type": "int" }
+  ]
+}''')
+    with pytest.raises(ValidationError) as e:
+        s.validate({'name': 'Obi-Wan'})
+    assert 'name' not in str(e)
+    assert 'age' in str(e)
+    with pytest.raises(ValidationError) as e:
+        s.validate({'name': 1, 'age': 2})
+    assert 'name' in str(e)
+    assert 'age' not in str(e)
+    s.validate({'name': 'Jar Jar', 'age': 42, 'sith': True})