# Coming from FastAPI

> Move work out of FastAPI request handlers and background tasks into durable Pyrula runs, with the run endpoints mounted on your app: a concept map, the scaffold and an executed example.

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

```bash
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](https://docs.pyrula.io/getting-started/existing-app.md#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`](https://docs.pyrula.io/reference/workflows.md#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](https://docs.pyrula.io/workflows/guides/http.md#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](https://docs.pyrula.io/workflows/guides/observability.md)) | Direct swap |
| `TestClient` | Unchanged for routes; `WorkflowTestEnv` for workflow logic | Direct swap |
| An agent API | `create_app(...)` from `pyrula.agents` ([HTTP](https://docs.pyrula.io/agents/serving/http.md)) | Small rewrite |

## Example

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

```python
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"]
```