Coming from Temporal

Pyrula has counterparts for Temporal’s workflow model: replay from history, activities as journaled steps, signals, queries, updates, child runs, continue-as-new, patching and schedules. There is no server or cluster; workers share a store and the journal lives there.

Scaffold

There is no pyrula init --temporal. Start from the workflow scaffold:

pyrula init --workflow
python app.py

Concept map

Temporal	Pyrula	Change
@workflow.defn class, @workflow.run	@workflow on an async def; parameters travel as JSON	Small rewrite
Activity, start_to_close_timeout	@step, or await ctx.step(name, fn); @step(timeout=) bounds one try	Small rewrite
Activity retry policy	RetryPolicy; Fatal and Poison never retry	Small rewrite
schedule_to_close, heartbeats	None; the worker renews the run’s lease itself	Not supported
Whole-workflow deadline	None; @workflow(timeout=) bounds one attempt	Not supported
workflow.sleep	await ctx.sleep(duration), ctx.sleep_until(dt)	Direct swap
Signal handler, wait_condition	await ctx.wait_for_signal(name, timeout=...); no predicate wait	Small rewrite
Signalling another workflow	ctx.send_signal(target_name, target_run_id, signal_name, payload)	Direct swap
@workflow.query	ctx.on_query(name, handler); handle.query(name) writes nothing	Small rewrite
@workflow.update, validator	ctx.on_update(name, handler, validator); the handler must not suspend	Small rewrite
Child workflow, ParentClosePolicy	ctx.run_child(name, params, parent_close="cancel"), "terminate" or "abandon"	Direct swap
Continue-as-new	await ctx.continue_as_new(params); ctx.continue_suggested at 80% of max_event_count_per_run	Direct swap
workflow.patched, deprecate_patch	await ctx.patched(id), await ctx.deprecate_patch(id)	Direct swap
Worker Versioning	@workflow(version=..., version_behavior="pinned" or "auto_upgrade") (Versioning)	Conceptual redesign
workflow.now(), uuid4(), random()	ctx.now() (Unix seconds as a float), ctx.uuid(), ctx.random()	Small rewrite
Workflow id reuse policy	submit(idempotency_key=...); a repeat returns the first run for 24 hours by default	Small rewrite
Cancel, terminate	handle.cancel() at the next step boundary; handle.terminate() now	Direct swap
Reset to an event	handle.retry() starts a new run that serves recorded steps; no event to reset to	Conceptual redesign
Schedules	Schedule.create(name, cron=... or every=..., timezone=, overlap=, catchup_window=); no backfill	Small rewrite
Task queues	None; a worker claims runs of the workflows it registers	Not supported
Search attributes, List Filter	submit(metadata=...); list_runs(name, owner=) has no field filter	Not supported
Time-skipping test environment	WorkflowTestEnv from pyrula.workflows.testing	Small rewrite
The replayer	assert_replay_compatible(workflow, history) and replay in CI	Direct swap
Nexus	None; use ctx.run_child or ctx.send_signal	Not supported
Server, cluster, UI	A shared store, pyrula dev, pyrula runs, Pyrula Cloud	Operational change

Example

One order run with a retried activity, a query, a signal, a patched branch and a one-day timer the test environment skips:

import asyncio
from datetime import timedelta

from pyrula.workflows import RetryPolicy, WorkflowContext, step, workflow
from pyrula.workflows.testing import WorkflowTestEnv

attempts: list[int] = []


# An activity with a retry policy.
@step(retry=RetryPolicy(max_attempts=3, initial_backoff=0.01, retryable=(ConnectionError,)))
async def reserve(sku: str) -> str:
    attempts.append(1)
    if len(attempts) == 1:
        raise ConnectionError("warehouse unreachable")
    return f"reserved {sku}"


@workflow(name="docs_from_temporal_order")
async def order(ctx: WorkflowContext, sku: str) -> dict:
    stage = {"now": "reserving"}
    ctx.on_query("stage", lambda: stage["now"])  # @workflow.query
    reserved = await reserve(sku)
    stage["now"] = "waiting for payment"
    paid = await ctx.wait_for_signal("paid", timeout=timedelta(days=2))  # @workflow.signal
    if await ctx.patched("label-v2"):  # workflow.patched
        label = await ctx.step("label_v2", lambda: f"v2 label for {sku}")
    else:
        label = await ctx.step("label", lambda: f"label for {sku}")
    await ctx.sleep(timedelta(days=1))  # workflow.sleep
    return {"reserved": reserved, "paid_by": paid.payload["by"], "label": label}


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.start(order, {"sku": "A-1"})
    await env.run_until_next_block(run)  # runs to the signal wait
    assert await run.query("stage") == "waiting for payment"
    await run.signal("paid", {"by": "card"})
    await env.run_until_complete(run)  # skips the one-day sleep
    assert await run.result(timeout=30) == {
        "reserved": "reserved A-1",
        "paid_by": "card",
        "label": "v2 label for A-1",
    }
    assert len(attempts) == 2


asyncio.run(main())

A child run is a separate run with its own journal, started by a worker or by the test environment:

@workflow(name="docs_from_temporal_ship")
async def ship(ctx: WorkflowContext, order_id: str) -> str:
    return await ctx.step("label", lambda: f"label for {order_id}")


@workflow(name="docs_from_temporal_fulfil")
async def fulfil(ctx: WorkflowContext, order_id: str) -> str:
    await ctx.wait_for_signal("paid")
    return await ctx.run_child(
        "docs_from_temporal_ship", {"order_id": order_id}, timeout=60, parent_close="abandon"
    )


async def fulfil_one() -> None:
    env = WorkflowTestEnv()
    run = await env.start(fulfil, {"order_id": "o-1"})
    await env.run_until_complete(run)  # parks on the signal wait
    await run.signal("paid", {})
    assert await run.result(timeout=30) == "label for o-1"


asyncio.run(fulfil_one())