Add one boundary to an existing app

Wrap one side-effecting function, such as charging a card, as a recorded step: its result is stored once, and a resumed run reads it back instead of calling it again. This page uses Python 3.11 to 3.15 and a POSIX shell, in an empty directory.

1. Install

pip install "pyrula[agents,cloud]"

2. Wrap the function

Save this as app.py. charge_card stands for your function.

app.py

import asyncio

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

CHARGES: list[str] = []  # stands in for the payment provider


# Your existing function. The decorator is the only change to it.
@step
def charge_card(order_id: str, cents: int) -> dict:
    CHARGES.append(order_id)
    return {"order_id": order_id, "charged": cents}


# The boundary: one workflow that calls it.
@workflow(name="checkout")
async def checkout(ctx, order_id: str) -> dict:
    return await charge_card(order_id, 4200)


async def main() -> None:
    runner = InlineWorkflowRunner(store=MemoryStore())
    run = await runner.client.submit("checkout", {"order_id": "A-1042"})
    print("result:", await run.result())
    print("charges:", CHARGES)
    for event in await run.events():
        print(event.kind)


if __name__ == "__main__":
    asyncio.run(main())

Run it on its own:

python app.py

result: {'order_id': 'A-1042', 'charged': 4200}
charges: ['A-1042']
run:init
step:pending
step:done
run:complete

step:done holds the result a resumed run reads back.

3. Run it under pyrula dev

pyrula dev app.py --params '{"order_id": "A-1042"}'

dev stays running. It prints the target, store and UI address, then the run:

Target: app.py:checkout
  charge_card  done
  completed - {'order_id': 'A-1042', 'charged': 4200}

Open the printed UI: address to see the events. With no Cloud key, dev prints Cloud: off (no PYRULA_CLOUD_API_KEY) - running locally only.

4. See it in Cloud

Stop dev, export the key from Setup, and start it again:

export PYRULA_CLOUD_API_KEY="<key>"
pyrula dev app.py --params '{"order_id": "A-1042"}'

dev prints Cloud: connected to ... and, once the first events are accepted, Cloud: run <run-id> -> <link>. The run’s charge_card step is there.

Start from a scaffold

pyrula init writes an app.py that runs offline, with no key or server, in the shape of your app:

Command	Writes
pyrula init --fastapi	A route that starts a durable checkout and returns its run id, and a RunRouter at /runs. A worker runs in-process; auth is off
pyrula init --celery	A welcome-email @workflow with a retried @step, and enqueue_welcome(user_id) submitting through the blocking client with delay= and an idempotency_key. A header comment maps each Celery concept to its counterpart
pyrula init --langgraph	A graph inside an @agent, with a scripted chat model in durable_chat_model and the tool’s effect in ctx.step. Needs pip install "pyrula[agents,langgraph,langchain]"
pyrula init --pydantic-ai	A @workflow running an Agent with capabilities=[PyrulaDurability(name=...)]. Needs pip install "pyrula[agents,pydantic-ai]" (Pydantic AI 2.62+). See Pydantic AI

Run the file directly, or under pyrula dev for the UI and, with PYRULA_CLOUD_API_KEY set, the Cloud link:

python app.py
pyrula dev app.py --open

The FastAPI scaffold also serves with uvicorn app:app.

Crash behavior

If the process dies after charge_card returned and its result was recorded, the resumed run doesn’t call it again.

Pyrula cannot make an external effect exactly-once. If the charge succeeded at the provider but the process died before the result was recorded, the resumed run calls charge_card again, so pass the provider an idempotency key derived from the order. pyrula init writes a crash_resume.py that shows a kill and resume.

From Django, Flask or Celery

Sync handlers and tasks use the blocking client in pyrula.workflows.sync. Create one per process at import time and share it between threads (From sync code).

from flask import Flask, jsonify
from celery import Celery
from pyrula.workflows.sync import WorkflowClient
from myapp.workflows import store

app = Flask(__name__)
celery = Celery("myapp")
client = WorkflowClient(store)


@app.post("/orders/<order_id>")
def place_order(order_id: str):
    run = client.submit("process_order", {"order_id": order_id}, idempotency_key=order_id)
    return jsonify(run_id=run.run_id), 202


@celery.task
def process_later(order_id: str) -> str:
    return client.submit("process_order", {"order_id": order_id}).run_id