Coming from FastAPI

Routes stay. Work that must finish even if the process dies (a charge, a provisioning job, anything handed to BackgroundTasks) moves into a Pyrula @workflow, and the route submits a run and returns its id. RunRouter serves run status, a server-sent events stream and control endpoints. A worker executes the runs: in the app’s lifespan in development, as its own process in production.

Scaffold

pyrula init --fastapi
python app.py        # starts one checkout in-process and prints the result
uvicorn app:app      # serves the app and the run endpoints

See Start from a scaffold.

Concept map

FastAPI	Pyrula	Change
BackgroundTasks.add_task	await client.submit(name, params) of a @workflow; the run survives a restart	Small rewrite
Handler with several outside calls	A @workflow with each call in a @step	Small rewrite
Handler returning the result	Return {"run_id": ...}; read the result from the run endpoints or a RunHandle	Conceptual redesign
app.include_router	RunRouter(...).attach(app, prefix="/runs")	Direct swap
Lifespan starting background work	WorkflowWorker in the lifespan for development; pyrula worker in production	Operational change
Idempotency-Key header	submit(idempotency_key=...); a retried POST gets the first run	Direct swap
Depends(...) auth	Yours for your routes; run endpoints take AuthConfig(provider=APIKeyAuth(...)) or JWTAuth (Authentication)	Small rewrite
WebSocket or SSE progress	GET /runs/{name}/stream/{run_id}; resumes from Last-Event-ID	Direct swap
Polling endpoint	GET /runs/{name}/runs/{run_id}	Direct swap
Cancel endpoint	DELETE /runs/{name}/stream/{run_id}; POST .../terminate ends it now	Direct swap
Callback route for a human decision	ctx.interrupt(...) answered on POST .../resume, or ctx.wait_for_signal(...) on POST .../signal	Conceptual redesign
asyncio.sleep in a background task	await ctx.sleep(...), a durable timer	Direct swap
asyncio.gather	await ctx.gather(...), each leg journaled	Direct swap
Retry decorators	@step(retry=RetryPolicy(...))	Direct swap
Request timeout	@step(timeout=) per try; @workflow(timeout=) per attempt (300 s)	Small rewrite
Rate-limit middleware	Yours for requests; @workflow(rate_limit=Rate(...)), concurrency=Limit(...) limit claims	Operational change
429 or 503 load shedding	RunRouterConfig(max_queue_depth=...) refuses a submit with Retry-After	Direct swap
Sync def route	The blocking pyrula.workflows.sync.WorkflowClient	Small rewrite
uvicorn --workers N	Size API and worker processes separately; WorkerOptions(max_concurrent_runs=)	Operational change
Admin page, log search	mount_run_ui(app, prefix="/ui", control=...), pyrula dev, Pyrula Cloud	Operational change
Error text in responses	Replaced unless RunRouterConfig(expose_error_details=True)	Operational change
OpenTelemetry instrumentation	Workflow spans and worker metrics (Observability)	Direct swap
TestClient	Unchanged for routes; WorkflowTestEnv for workflow logic	Direct swap
An agent API	create_app(...) from pyrula.agents (HTTP)	Small rewrite

Example

The scaffold’s route, driven through TestClient. A retried POST gets the same run, and the charge happens once:

import asyncio
import tempfile
import time
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pyrula.contracts.web import AuthConfig
from pyrula.workflows import (
    WorkerOptions,
    WorkflowContext,
    WorkflowWorker,
    open_store,
    step,
    workflow,
)
from pyrula.workflows.web import RunRouter, RunRouterConfig, StoreControl

charges: list[str] = []


@step
def charge_card(order_id: str) -> dict:
    charges.append(order_id)
    return {"order_id": order_id, "charged": 4200}


# What used to be a BackgroundTasks function.
@workflow(name="docs_from_fastapi_checkout")
async def checkout(ctx: WorkflowContext, order_id: str) -> dict:
    return await charge_card(order_id)


# A fresh file per run of this example; an app keeps one store path for good.
store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/checkout.db")
worker = WorkflowWorker(
    workflows=[checkout],
    store=store,
    options=WorkerOptions(claim_poll_interval=0.05, timer_poll_interval_s=0.1),
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(worker.run())  # development: the worker in the app's process
    try:
        yield
    finally:
        worker.stop()
        await task


app = FastAPI(lifespan=lifespan)
# provider=None switches authentication off, for local work only.
RunRouter(
    control=StoreControl(store),
    config=RunRouterConfig(auth_config=AuthConfig(provider=None)),
    workflows=[checkout],
).attach(app, prefix="/runs")


@app.post("/orders/{order_id}/checkout", status_code=202)
async def start_checkout(order_id: str) -> dict:
    run = await worker.client.submit(
        "docs_from_fastapi_checkout",
        {"order_id": order_id},
        idempotency_key=f"checkout:{order_id}",
    )
    return {"run_id": run.run_id}


with TestClient(app) as http:
    first = http.post("/orders/A-1/checkout").json()
    assert http.post("/orders/A-1/checkout").json() == first  # a retried POST
    status_url = f"/runs/docs_from_fastapi_checkout/runs/{first['run_id']}"
    deadline = time.monotonic() + 30
    while http.get(status_url).json()["status"] != "complete":
        assert time.monotonic() < deadline
        time.sleep(0.1)
    assert charges == ["A-1"]