Quickstart

Install "pyrula[workflows]", then scaffold a runnable workflow:

pip install "pyrula[workflows]"
pyrula init --workflow
pyrula dev app.py --open

Until the first public release the packages need a source build; see Install.

Generated workflow

pyrula init --workflow writes app.py: reserve inventory, charge a card, sleep, send a receipt. The getting-started quickstart shows the file and its output.

Send this to Cloud

Export the key from Cloud’s Setup and restart dev. It prints Cloud: connected to ... and a Cloud: run <run-id> -> <link> line.

export PYRULA_CLOUD_API_KEY="<key>"
pyrula dev app.py

See Connect to Pyrula Cloud for the full output.

Minimal workflow

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


@step
def build(name: str) -> str:
    return f"hello, {name}"


@workflow(name="greet")
async def greet(ctx, name: str) -> str:
    return await build(name)


async def main() -> None:
    runner = InlineWorkflowRunner(store=MemoryStore())
    run = await runner.client.submit("greet", {"name": "ada"})
    print(await run.result())        # hello, ada
    for event in await run.events():
        print(event.kind, event.payload)


asyncio.run(main())

submit queues, result() runs it in this process, events() returns the journal. MemoryStore is for tests; SQLiteStore("pyrula.db") and Valkey survive restarts (Choosing a store).

Parameters

Params are validated against the body’s annotations, and a dataclass, Pydantic model or list[Order] arrives as that type. Bad or missing input ends the run with run:error code invalid_request (the error never echoes the input). An annotation that can’t become a schema raises TypeError at definition. Concurrency and rate keys read the raw submitted JSON, before coercion.

Options

@workflow(
    name="report",
    timeout=600.0,                  # per attempt, default 300
    max_event_count_per_run=10_000,     # default 50,000
    max_event_payload_bytes=1_000_000,  # the default; None lifts the limit
    max_attempt_wall_seconds=3600.0,    # min'd with timeout, still per attempt
)
async def report(ctx):
    ...

A run with more than max_event_count_per_run events ends the same way, its reason pointing at ctx.continue_as_new; the default is 50,000 and None declares none, and a store that trims a run’s stream (ValkeyStore, below its maxlen) lowers it to what the store keeps either way. ctx.continue_suggested turns true at 80% of it. A payload over max_event_payload_bytes (a step result, submit params) ends the run with stream:overflow then run:error (stream_too_large) naming the step. An oversized submit is refused with PayloadTooLarge before anything is written. Don’t catch it in the body: once a cap trips, later cap checks for that attempt are off.

Next: Runs as Event Streams.