# Coming from Temporal

> Map Temporal workflows, activities, signals, queries, updates, child workflows, patching and schedules to Pyrula: a concept map and an executed example.

Source: https://docs.pyrula.io/getting-started/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:

```bash
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`](https://docs.pyrula.io/reference/workflows.md#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](https://docs.pyrula.io/workflows/guides/versioning.md)) | 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](https://docs.pyrula.io/workflows/guides/replay-ci.md) | 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:

```python
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:

```python
@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())
```