How to catch faust python json serialization error?

Question

I have simple faust agent. It consumes jsons from kafka topic, and parses them to dicts by default faust serializator:

@app.agent(source_topic, sink=[destination_topic])
async def fetch(records):
    async for record in records:
        result = do_some_stuff(record)
        yield result

The deserialization itself happens somewhere outside my code, it managed by faust framework, not by me. How I can catch and process deserialization exceptions, e.g in case of invalid json?

Answer 1

You could manually serialize your data and catch your error:

topic = app.topic('custom', value_type=bytes)

@app.agent
async def processor(stream):
    async for payload in stream:
        data = json.loads(payload)

source: https://faust.readthedocs.io/en/latest/userguide/models.html