Merge pull request #4 from DisasterAWARE/rename-serde

Rename SchemaRegistry(De)Serializer

Author: cdebost

README.md

@@ -67,15 +67,15 @@ from aws_schema_registry.avro import AvroSchema
 # In this example we will use kafka-python as our Kafka client,
 # so we need to have the `kafka-python` extras installed and use
 # the kafka adapter.
-from aws_schema_registry.adapter.kafka import SchemaRegistrySerializer
+from aws_schema_registry.adapter.kafka import KafkaSerializer
 from kafka import KafkaConsumer
 
 # Create the schema registry client, which is a façade around the boto3 glue client
 client = SchemaRegistryClient(glue_client,
                               registry_name='my-registry')
 
 # Create the serializer
-serializer = SchemaRegistrySerializer(client)
+serializer = KafkaSerializer(client)
 
 # Create the producer
 producer = KafkaProducer(value_serializer=serializer)
@@ -91,7 +91,7 @@ data = {
     'favorite_number': 6
 }
 producer.send('my-topic', value=(data, schema))
-# the value MUST be a tuple when we're using the SchemaRegistrySerializer
+# the value MUST be a tuple when we're using the KafkaSerializer
 ```
 
 Read Kafka messages with `SchemaRegistryDeserializer`:
@@ -102,15 +102,15 @@ from aws_schema_registry import SchemaRegistryClient
 # In this example we will use kafka-python as our Kafka client,
 # so we need to have the `kafka-python` extras installed and use
 # the kafka adapter.
-from aws_schema_registry.adapter.kafka import SchemaRegistryDeserializer
+from aws_schema_registry.adapter.kafka import KafkaDeserializer
 from kafka import KafkaConsumer
 
 # Create the schema registry client, which is a façade around the boto3 glue client
 client = SchemaRegistryClient(glue_client,
                               registry_name='my-registry')
 
 # Create the deserializer
-deserializer = SchemaRegistryDeserializer(client)
+deserializer = KafkaDeserializer(client)
 
 # Create the consumer
 consumer = KafkaConsumer('my-topic', value_deserializer=deserializer)

src/aws_schema_registry/__init__.py

@@ -3,21 +3,19 @@
 from .schema import (
     CompatibilityMode, DataFormat, Schema, SchemaVersion, ValidationError
 )
-from .serde import (
-    DataAndSchema, SchemaRegistryDeserializer, SchemaRegistrySerializer
-)
+from .serde import DataAndSchema, KafkaDeserializer, KafkaSerializer
 
 __version__ = '1.0.0rc5'
 
 __all__ = [
     'CompatibilityMode',
     'DataAndSchema',
     'DataFormat',
+    'KafkaDeserializer',
+    'KafkaSerializer',
     'Schema',
     'SchemaRegistryClient',
-    'SchemaRegistryDeserializer',
     'SchemaRegistryException',
-    'SchemaRegistrySerializer',
     'SchemaVersion',
     'ValidationError'
 ]

src/aws_schema_registry/adapter/kafka.py

@@ -6,22 +6,22 @@
 from kafka import Serializer, Deserializer
 
 from aws_schema_registry import (
-    SchemaRegistrySerializer as _SchemaRegistrySerializer,
-    SchemaRegistryDeserializer as _SchemaRegistryDeserializer
+    KafkaSerializer as _KafkaSerializer,
+    KafkaDeserializer as _KafkaDeserializer
 )
 
 
-class SchemaRegistrySerializer(Serializer):
+class KafkaSerializer(Serializer):
     def __init__(self, *args, **kwargs):
-        self._serializer = _SchemaRegistrySerializer(*args, **kwargs)
+        self._serializer = _KafkaSerializer(*args, **kwargs)
 
     def serialize(self, topic, value):
         return self._serializer.serialize(topic, value)
 
 
-class SchemaRegistryDeserializer(Deserializer):
+class KafkaDeserializer(Deserializer):
     def __init__(self, *args, **kwargs):
-        self._deserializer = _SchemaRegistryDeserializer(*args, **kwargs)
+        self._deserializer = _KafkaDeserializer(*args, **kwargs)
 
     def deserialize(self, topic, bytes_):
         return self._deserializer.deserialize(topic, bytes_)

src/aws_schema_registry/serde.py

@@ -2,15 +2,9 @@
 
 import functools
 import logging
-import sys
 from typing import Any, NamedTuple
 from uuid import UUID
 
-if sys.version_info[1] < 8:  # for py37
-    from typing_extensions import Protocol
-else:
-    from typing import Protocol
-
 from aws_schema_registry.avro import AvroSchema
 from aws_schema_registry.client import SchemaRegistryClient
 from aws_schema_registry.codec import decode, encode, UnknownEncodingException
@@ -33,15 +27,7 @@ class DataAndSchema(NamedTuple):
     schema: Schema
 
 
-class Serializer(Protocol):
-    def serialize(self, topic: str, record: DataAndSchema): ...
-
-
-class Deserializer(Protocol):
-    def deserialize(self, topic: str, bytes_: bytes): ...
-
-
-class SchemaRegistrySerializer:
+class KafkaSerializer:
     """Kafka serializer that uses the AWS Schema Registry.
 
     Arguments:
@@ -94,7 +80,7 @@ def _get_schema_version(self, topic: str, schema: Schema) -> SchemaVersion:
         )
 
 
-class SchemaRegistryDeserializer:
+class KafkaDeserializer:
     """Kafka serializer that uses the AWS Schema Registry.
 
     Arguments:
@@ -106,14 +92,16 @@ class SchemaRegistryDeserializer:
         secondary_deserializer: optional deserializer to pass through
             to when processing values with an unrecognized encoding.
             This is primarily use to migrate from other schema
-            registries or handle schema-less data.
+            registries or handle schema-less data. The secondary deserializer
+            should either be a callable taking the same arguments as
+            deserialize or an object with a matching deserialize method.
     """
 
     def __init__(
         self,
         client: SchemaRegistryClient,
         return_record_name: bool = False,
-        secondary_deserializer: Deserializer = None
+        secondary_deserializer=None
     ):
         self.client = client
         self.return_record_name = return_record_name
@@ -126,6 +114,8 @@ def deserialize(self, topic: str, bytes_: bytes):
             data_bytes, schema_version_id = decode(bytes_)
         except UnknownEncodingException as e:
             if self.secondary_deserializer:
+                if callable(self.secondary_deserializer):
+                    return self.secondary_deserializer(topic, bytes_)
                 return self.secondary_deserializer.deserialize(topic, bytes_)
             else:
                 raise SchemaRegistryException(

tests/integration/java/test_java_integration.py

@@ -5,7 +5,7 @@
 from aws_schema_registry import DataAndSchema, SchemaRegistryClient
 from aws_schema_registry.avro import AvroSchema
 from aws_schema_registry.adapter.kafka import (
-    SchemaRegistryDeserializer, SchemaRegistrySerializer
+    KafkaDeserializer, KafkaSerializer
 )
 
 LOG = logging.getLogger(__name__)
@@ -23,8 +23,8 @@
 
 def test_interop_with_java_library(glue_client, registry, boto_session):
     client = SchemaRegistryClient(glue_client, registry_name=registry)
-    serializer = SchemaRegistrySerializer(client)
-    deserializer = SchemaRegistryDeserializer(client)
+    serializer = KafkaSerializer(client)
+    deserializer = KafkaDeserializer(client)
 
     data = {
         'name': 'John Doe',

tests/integration/kafka/test_kafka_integration.py

@@ -8,7 +8,7 @@
 from aws_schema_registry import DataAndSchema, SchemaRegistryClient
 from aws_schema_registry.avro import AvroSchema
 from aws_schema_registry.adapter.kafka import (
-    SchemaRegistryDeserializer, SchemaRegistrySerializer
+    KafkaDeserializer, KafkaSerializer
 )
 from aws_schema_registry.naming import record_name_strategy
 
@@ -62,10 +62,10 @@ def test_produce_consume_with_ser_de_schema_registry(
     client = SchemaRegistryClient(
         glue_client, registry_name=registry
     )
-    serializer = SchemaRegistrySerializer(
+    serializer = KafkaSerializer(
         client, schema_naming_strategy=record_name_strategy
     )
-    deserializer = SchemaRegistryDeserializer(client)
+    deserializer = KafkaDeserializer(client)
 
     producer = KafkaProducer(
         value_serializer=serializer,