faust-manual-seralization.py

import fastavro
import json

json_stream = faust_app.topic('my-json-topic', value_type=bytes)

# manual deseralization of JSON
@app.agent(json_stream)
async def processor(records):
    async for record in records:
        des_data = json.loads(record)

# manual deseralization of AVRO
avro_stream = faust_app.topic('my-avro-topic', value_type=bytes)
avro_schema_str = {
    'doc': 'A schema',
    'name': 'Transaction',
    'namespace': 'test',
    'type': 'record',
    'fields': [
        {'name': 'amount', 'type': 'float'},
        {'name': 'receiver', 'type': 'str'},
        {'name': 'date', 'type': 'str'},
    ],
}

def fast_avro_decode(schema, encoded_message):
    stringio = io.BytesIO(encoded_message)
    return fastavro.schemaless_reader(stringio, schema)

@app.agent(avro_stream)
async def processor(records):
    async for record in records:
        schema = fastavro.parse_schema(avro_schema_str)
        des_data = fast_avro_decode(schema, record)