Starting and talking to runs

Workers execute runs. WorkflowClient starts them and talks to them. It needs only the store the workers use, so it can live in a web handler, a script or another service.

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)

submit returns a RunHandle. Keep it, or rebuild one later from its id with client.handle(name, run_id). Every method is async, and every handle method follows a continue_as_new chain to its live run.

RunHandle method	Does	HTTP route
signal(signal, payload, *, signal_id)	Wake ctx.wait_for_signal	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)	Apply ctx.on_update and return its value	POST .../update
resume(action, value, *, wait_id)	Answer ctx.interrupt	POST .../resume
cancel()	Stop the run at its next step	DELETE /{name}/stream/{run_id}
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)	Queue a run; returns its handle	POST /{name}/runs
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.

• submit with idempotency_key returns the first run on a repeat. owner tags the tenant for fair claims. The workflow’s version and flow control are stamped from its decorator, so import the module that defines it; submitting a name this process hasn’t imported works but skips both, and logs it.
• 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.

Errors

Exception	When
RunNotFoundError	No such run
RunFinishedError	signal or update on a finished run
RunNotWaitingError	resume when the run isn’t waiting on that interrupt
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

A worker exposes one as worker.client. Its submit raises BackpressureRejected while the worker’s cloud egress is behind. A complete program, worker and client in one process:

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=10) == ["A-1"]

    worker.stop()
    await loop


asyncio.run(main())

Without a worker

InlineWorkflowRunner and the time-skipping WorkflowTestEnv have a client too. Their handles run in your process: result, update and signal drive the run there instead of waiting for a worker, so the same code works in a test and in production.

WorkflowTestEnv (from pyrula.workflows.testing) also keeps a virtual clock, so a test of a month-long workflow finishes at once:

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, and env.run_until_complete(run) drives a run until it ends or waits on a signal.