I've implemented a Kafka topic that tries to buffer multiple events for efficiency but when I try and modify a Faust table I get the error message Cannot modify table key from outside of stream iteration
class BroadcastCounts(faust.Record):
delivers: int
bounces: int
opens: int
clicks: int
unsubscribes: int
broadcast_counts = app.Table('broadcast_counts', value_type=BroadcastCounts)
@app.agent(broadcast_event_topic)
async def broadcast_event(stream):
async for events in stream.take(1000, within=10):
broadcast_counts['f7fb7078-d442-4219-8494-bfa5ab532230'] = BroadcastCounts(
delivers=0,
bounces=0,
opens=0,
clicks=0,
unsubscribes=0,
)
After some testing it seems that doing a
take(...)
is not part of a stream so you can't access tables.