Streaming Sources

ctx.stream pulls batches from a Source and tracks offsets through the run stream. After a crash, replay skips batches whose offsets are recorded as committed.

from pyrula.workflows import workflow

@workflow(name="ingest")
async def ingest(ctx):
    async for batch in ctx.stream(source, timeout_ms=5000):
        for record in batch.records:
            await handle(record)
        await ctx.commit(source, batch)

ctx.stream yields a Batch at a time. ctx.commit persists that batch’s offset. Replay skips committed batches. A crash before ctx.commit can repeat the batch’s handling; use idempotency keys for external effects.

A Batch carries its position and payload:

from pyrula.workflows import Batch   # source_id, partition, offset, records

Sources

A Source yields batches and knows how to seek back to an offset on resume. Implement the protocol for your transport, or use InMemorySource in tests:

from pyrula.workflows import InMemorySource

source = InMemorySource(source_id="orders", records=[...], batch_size=10)

For Kafka specifically, reach for pyrula.kafka directly. This source abstraction is for wiring arbitrary inputs into a durable run with the same offset-commit guarantee.

Sinks

Write outputs inside ctx.step; there is no Sink protocol.