Starting and talking to runs

from pyrula.workflows import ValkeyStore, WorkflowClient

client = WorkflowClient(ValkeyStore(url="valkey://localhost:6379"))

run = await client.submit("charge", {"amount_cents": 500}, idempotency_key="order-7")
result = await run.result(timeout=60)

The client needs only the workers’ store. submit returns a RunHandle; rebuild one with client.handle(name, run_id). Everything is async and follows a continue_as_new chain to the live run. A retried signal_id or update_id stays idempotent across hops for completed_ttl (a day by default on Valkey).

RunHandle method	Does	HTTP route
signal(signal, payload, *, signal_id)	Wake ctx.wait_for_signal; returns "delivered" or "buffered"	POST .../runs/{run_id}/signal
query(query, *args, **kwargs)	Read state through ctx.on_query; writes nothing	POST .../query
update(update, payload, *, update_id, timeout=30.0)	Apply ctx.on_update and return its value	POST .../update
resume(action, value, *, wait_id)	Answer ctx.interrupt	POST .../resume
cancel(*, reason)	Request cancellation and queue parked runs for cleanup	DELETE /{name}/stream/{run_id}
terminate(*, reason)	End the run now with no worker; returns TerminateResult	POST .../runs/{run_id}/terminate
retry()	New run that reuses a failed run’s recorded steps (Replay and recovery)	-
status()	Current RunStatus	GET .../runs/{run_id}
result(*, timeout, on_interrupt)	Wait for the run to finish	-
events()	The run’s events, oldest first	GET .../runs/{run_id}/events

WorkflowClient method	Does	HTTP route
submit(name, params, *, idempotency_key, owner, metadata, allow_unregistered=True, start_at, delay)	Queue a run, optionally for later; returns its handle	POST /{name}/runs (takes no start time yet)
trigger(name, event)	Feed a Collect/Debounce buffer; returns a handle when a batch fires	POST /{name}/events
handle(name, run_id)	A handle to an existing run	-

The client also has each handle method with name, run_id in front, for one-off calls.

result(on_interrupt=handler) calls handler(payload, run_id) for each ctx.interrupt and resumes with the return value: an InterruptResponse(action, value) with action one of approve, reject, respond or edit, or any other value as a respond answer. An unknown action raises ValueError and leaves the run suspended. On an expects="approval" interrupt, the bare string "approve" or "reject" is shorthand for that action.

• submit with idempotency_key returns the first run on a repeat. owner tags the tenant for fair claims. Version and flow control are stamped from the decorator. A process that submits by name without importing the module gets no version pin, and takes the Limit and Rate a serving worker recorded in the store (reread at most every 10 seconds).
• start_at and delay hold the body back (Starting a run later).
• update waits for whichever worker applies it. Past timeout it raises UpdateTimeout; the update stays queued, so retry with the same update_id.
• resume answers the interrupt the run is waiting on unless you pass wait_id.

Ids in a long chain

Keep the start id submit returned. It reaches the live run while the chain is open and for completed_ttl after; an id the chain has moved past works for at least completed_ttl after its successor took over, then raises RunNotFoundError. ctx.run_id is the current run’s id, so a body that needs a handle to itself carries the start id in its params.

Before a run is submitted, only cancel and pause are held. An update to such an id raises RunNotFoundError; a signal is refused.

Errors

Exception	When
RunNotFoundError	No such run, or an update to an id that was only cancelled or paused before it was submitted
TerminateUnsupported	terminate on a store without the terminate capability
RunFinishedError	signal or update on a finished run; signal also once a cancel is requested
RunNotWaitingError	resume when the run isn’t waiting on that interrupt
RunNotStartedError	update to a run still waiting for its start_at or delay
IdempotencyConflict	submit reusing an idempotency_key with a different start_at or delay
QueryError	Unknown query name
UpdateRejected / UpdateError / UpdateTimeout	Validator refused / handler raised / no worker applied it in time
DurableInvocationError / DurableInvocationCancelled / DurableInvocationTimeout	result on a failed / cancelled / unfinished run

All are importable from pyrula.workflows.

In the same process

worker.client is a client on the worker’s store. Its submit raises BackpressureRejected while the worker’s Cloud egress is behind.

import asyncio
from pyrula.workflows import MemoryStore, WorkerOptions, WorkflowWorker, workflow


@workflow(name="cart")
async def cart(ctx) -> list[str]:
    items: list[str] = []

    async def add(sku: str) -> int:
        items.append(sku)
        return len(items)

    ctx.on_update("add", add)
    ctx.on_query("items", lambda: items)
    await ctx.wait_for_signal("checkout")
    return items


async def main() -> None:
    worker = WorkflowWorker(
        workflows=[cart], store=MemoryStore(), options=WorkerOptions(claim_poll_interval=0.05)
    )
    loop = asyncio.create_task(worker.run())
    client = worker.client

    run = await client.submit("cart")
    while await run.status() != "signal_waiting":
        await asyncio.sleep(0.05)
    assert await run.update("add", {"sku": "A-1"}) == 1
    assert await run.query("items") == ["A-1"]
    await run.signal("checkout")
    assert await run.result(timeout=30) == ["A-1"]

    worker.stop()
    await loop


asyncio.run(main())

From sync code

pyrula.workflows.sync has a blocking WorkflowClient and RunHandle for Django and Flask views, Celery tasks and RQ jobs. They are named like the async ones, as with httpx.Client and httpx.AsyncClient. Workflow bodies stay async def; only the caller changes.

import asyncio
import threading

from pyrula.workflows import MemoryStore, WorkerOptions, WorkflowWorker, workflow
from pyrula.workflows.sync import WorkflowClient


@workflow(name="sync_greeting")
async def sync_greeting(ctx, who: str) -> str:
    return await ctx.step("greet", lambda: f"hello {who}")


store = MemoryStore()
worker = WorkflowWorker(
    workflows=[sync_greeting], store=store, options=WorkerOptions(claim_poll_interval=0.05)
)
threading.Thread(target=lambda: asyncio.run(worker.run()), daemon=True).start()

with WorkflowClient(store) as client:
    run = client.submit("sync_greeting", {"who": "ada"})
    assert run.result(timeout=30) == "hello ada"
    rows, _ = client.list_runs("sync_greeting")
    assert [row["run_id"] for row in rows] == [run.run_id]

Create one client per process, at import time, and reuse it. It owns one event loop on a daemon thread, started on the first call, and every thread of a threaded server can share it. Errors are the async client’s own: result(timeout=) raises DurableInvocationTimeout.

The methods are the async client’s (signatures in the sync client reference), plus list_runs, which calls the store’s own list_runs on the calling thread. The async constructor’s admit= hook is not offered.

• A call cut short may still have taken effect. Ctrl+C while waiting re-raises KeyboardInterrupt, and close() from another thread makes a waiting call raise RuntimeError. After either, retry with the idempotency parameters: idempotency_key on submit, signal_id on signal, update_id on update.
• result(on_interrupt=) handlers run on the client’s loop thread, shared by every caller. A handler that blocks stalls them all.
• close() stops the thread and is safe to call twice; with calls it for you. A call after it raises RuntimeError("client is closed"). A client left open is closed at interpreter exit.
• A sync method called from a thread running an event loop raises RuntimeError. Use pyrula.workflows.WorkflowClient there.
• A client created before os.fork() (Celery prefork, gunicorn with a preloaded app) builds its own loop and thread in the child. Fork while no thread is inside a store call, and open stores whose sockets don’t survive a fork (Valkey) after it. SQLiteStore reopens its connections per process, but a fork taken mid-call has failed in the child with disk I/O error.

Starting a run later

delay (seconds or timedelta) or start_at (timezone-aware datetime) holds the body until then. One of them, not both.

import asyncio
from datetime import timedelta

from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv


@workflow(name="send_receipt")
async def send_receipt(ctx, order: str) -> str:
    return f"receipt for {order}"


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.client.submit("send_receipt", {"order": "o-1"}, delay=timedelta(hours=2))
    assert await run.result() == "receipt for o-1"   # the two hours are virtual


asyncio.run(main())

from datetime import UTC, datetime, timedelta

await client.submit("send_receipt", {"order": "o-1"},
                    start_at=datetime.now(UTC) + timedelta(days=30))

The wait is the run’s first sleep: no worker or concurrency slot held, and it survives restarts. One worker claims the run briefly to arm it. It reads pending until armed, then sleeping. A delay becomes an absolute time on the client’s clock at submit.

• Bounds. A negative, NaN, infinite or boolean delay, a naive start_at, or a time more than 365 days out raises ValueError. A past start_at or zero delay starts now.
• Lateness. The body can start late by timer_poll_interval_s plus the claim.
• Idempotency. Same idempotency_key and same timing returns the first run; different timing raises IdempotencyConflict. The key lives as long as the delay.
• Before it starts. cancel and terminate end the run without running the body, and a pause keeps it from beginning. Signals are buffered. An update raises RunNotStartedError (409 over HTTP).
• Retries and successors. A retry does not inherit the delay, and a continue_as_new successor starts at once. Schedules and Kafka records take no start time.
• Deploys. A newer worker can adopt a waiting run if the run follows versions (auto_upgrade, the default when a version is set); a pinned run waits for its version.

Without a worker

InlineWorkflowRunner and WorkflowTestEnv have a client whose handles drive the run in your process. WorkflowTestEnv (from pyrula.workflows.testing) keeps a virtual clock and skips to a sleep’s wake time and to the deadline of a wait_for_signal or interrupt with timeout=. A wait with no deadline stays parked until the test answers it.

import asyncio
from datetime import timedelta

from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv


@workflow(name="trial")
async def trial(ctx, user: str) -> str:
    await ctx.sleep(timedelta(days=30))
    return f"{user}: trial over"


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.start(trial, {"user": "ada"})
    assert await run.result() == "ada: trial over"   # the 30 days are virtual


asyncio.run(main())

await env.advance(timedelta(...)) moves the clock by hand; env.run_until_complete(run) drives a run until it ends or waits on a signal.

A child started with ctx.run_child runs in the test env while its parent waits on it, and run_child(timeout=) counts virtual time. env.children(run) returns the handles of the children a run waits on. Signal a parked child through its handle from another task; if every child is parked on input and nothing changes for two seconds, the wait raises a RuntimeError naming each child and its signal.

Cancellation

Cancel is graceful and needs a worker; to end a run now without one, see Cancel and terminate.

await client.cancel(name, run_id) records a request. A sleeping, signal-waiting, interrupted or paused run is queued for its cancelled terminal with no further work. A running step finishes, and the next live effect reads the store before starting. Recorded results still replay.

Once cancel is requested, signal raises RunFinishedError (409 over HTTP). Cleanup goes through normal version, tenant, concurrency and rate admission, so a saturated queue or missing worker delays it; a parked run can show in_progress until run:error with code="cancelled" arrives.

Inside a body, ctx.cancelled is a hint refreshed by heartbeats (10 s default) and can be stale during outages. Open child runs follow ctx.run_child’s parent_close (cancelled by default; see Parallel steps and child runs). A cancelled child surfaces to its parent as ChildWorkflowError.