avro_schema_decoder.py

import io
import json
import faust
import fastavro


class AvroSchemaDecoder(faust.Schema):
  
    """An extension of Faust Schema class. The class is used by Faust when
    creating streams from Kafka topics. The decoder deserializes each message 
    according to the AVRO schema injected in each message's header.
    """
    
    def __fast_avro_decode(self, schema, encoded_message):
        stringio = io.BytesIO(encoded_message)
        return fastavro.schemaless_reader(stringio, schema)

    def loads_value(self, app, message, *, loads=None, serializer=None):
        headers = dict(message.headers)
        avro_schema = fastavro.parse_schema(json.loads(headers["avro.schema"]))
        return self.__fast_avro_decode(avro_schema, message.value)

    def loads_key(self, app, message, *, loads=None, serializer=None):
        return json.loads(message.key)
      
      
stream = faust_app.topic(
    "my-topic",
    value_type=bytes,
    key_type=str,
    schema=AvroSchemaDecoder(),
)