Quickstart

pip install pyrula-workflows, then define a @workflow. The body takes a context (ctx) and uses ctx for anything with a side effect, so it replays cleanly.

import asyncio
from pyrula.workflows import workflow, InlineWorkflowRunner, MemoryStore


@workflow(name="greet")
async def greet(ctx, name: str) -> str:
    # ctx.step runs the work once and replays the recorded result on recovery
    return await ctx.step("build", lambda: f"hello, {name}")


async def main() -> None:
    runner = InlineWorkflowRunner(store=MemoryStore())
    run_id = runner.submit("greet", params={"name": "ada"})
    for event in await runner.run("greet", run_id) or []:
        print(event.kind, event.payload)


asyncio.run(main())

submit enqueues a run and hands back its id. run executes it and appends lifecycle and step events to the run’s stream. MemoryStore keeps that stream in process, which is what you want for tests. Swap in a Valkey or Redis store to survive restarts.

The thing to internalize: side effects go through ctx.step. On a replay the step is skipped and its recorded result comes back, so the work doesn’t run twice. See Determinism for why that matters.

Options

@workflow takes a few caps. timeout bounds one execution attempt; the rest bound the run’s persisted event stream so a runaway workflow can’t grow it without limit.

@workflow(
    name="report",
    timeout=600.0,                  # wall-clock cap per attempt
    max_event_count_per_run=10_000,
    max_event_payload_bytes=1_000_000,
    max_run_wall_seconds=3600.0,    # cap across all attempts
)
async def report(ctx):
    ...

Next: Runs as Event Streams.