Streaming Sources

ctx.stream pulls batches from a Source and tracks offsets through the run stream, so a crash resumes at the last committed offset instead of reprocessing. It’s how a workflow consumes a queue or a topic without double-handling records on replay.

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 any batch whose offset is already committed, which is what makes consumption exactly-once across a crash.

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

There is no Sink counterpart here, by design. Writing a result out is just another durable step (ctx.step), and the batch-to-warehouse sink machinery lives in pyrula.pipelines. This guide is about pulling an input into a durable run, not about output adapters.