FastAPI: Project: a production-grade orders API

Every part of this guide, in one service a senior team would be comfortable putting in front of customers. Brew & Bean is a chain of coffee bars: customers order from an app, baristas move orders through the kitchen, and a delivery partner receives webhooks. The API handles retries without duplicate orders, concurrent edits without lost updates, customers who try other people’s order numbers, a partner whose endpoint goes down, and a deploy in the middle of it all. The code is in fastapi/project/.

1. What it does, and where you learned it

2. How a request flows

fastapi/project/

orders_api/
  main.py            create_app(): lifespan, middleware, routers
  config.py          Settings (ORDERS_* variables) with production guards
  db.py              engine, session-per-request dependency, UTCDateTime
  models.py          User, Product, Order, OrderItem, IdempotencyRecord, OutboxEvent
  schemas.py         camelCase API models, separate In / Patch / Out
  security.py        bcrypt, JWT, current_user, require_admin
  services.py        ownership, totals, status machine, idempotency, ETags, cursors
  outbox.py          outbox events, signing, relay_once / relay_forever
  events.py          in-process pub/sub for SSE
  ratelimit.py       sliding-window limiter
  errors.py          ApiError and RFC 9457 handlers
  observability.py   JSON logs, request ids, access log
  seed.py            demo accounts and menu (dev only)
  routers/           auth.py · products.py · orders.py · health.py
tests/               conftest.py and six test modules
demo.py              end-to-end demo on real ports;  demo_receiver.py  the partner's endpoint
Dockerfile · docker-compose.yml · requirements.txt · .env.example · README.md

3. Run it

from the repository root

cd fastapi/project
pip install -r requirements-dev.txt
python demo.py                                           # two real servers, the whole story
python -m pytest                                         # 51 async tests
uvicorn orders_api.main:create_app --factory --reload    # then open http://127.0.0.1:8000/docs

python demo.py (real output)

== 1. Started: API and webhook receiver on real ports
  GET /readyz -> {'status': 'ready', 'database': 'ok'}

== 2. Log in (OAuth2 password flow, JWT access tokens)
  [email protected]   role=customer token eyJhbGciOiJIUzI1…  expires in 900 s  (bcrypt check: 0.46 s)
  [email protected]   role=customer token eyJhbGciOiJIUzI1…  expires in 900 s  (bcrypt check: 0.45 s)
  [email protected]  role=admin    token eyJhbGciOiJIUzI1…  expires in 900 s  (bcrypt check: 0.45 s)

== 3. Place an order with an Idempotency-Key, then retry it
  menu: 7 products, Cache-Control: public, max-age=60
  POST /v1/orders       -> 201  Location /v1/orders/1  ETag "1.1"  total ₹450.00
  network retry, same key -> 201  Idempotent-Replayed: true  same order: True
  5 simultaneous clicks   -> statuses [201], distinct orders created: 1
  same key, other body    -> 422  idempotency-key-reused

== 4. Object-level authorization
  ravi GET asha's order          -> 404 (not 403: no hint it exists)
  ravi sets a status (admin only) -> 403
  asha sends her own price        -> 422 Extra inputs are not permitted: totalMinor

== 5. Conditional requests: ETag and If-Match
  GET with If-None-Match "1.1"  -> 304
  PATCH without If-Match      -> 428
  PATCH with If-Match "1.1"   -> 200  new ETag "1.2"
  PATCH with stale "1.1"      -> 412  the order was changed by someone else; GET it again and retry

== 6. Live status over Server-Sent Events while the kitchen works
  t=0.01s  SSE event: status=pending    version=2
  t=0.39s  SSE event: status=paid       version=3
  t=0.78s  SSE event: status=preparing  version=4
  t=1.18s  SSE event: status=ready      version=5
  t=1.51s  SSE event: status=completed  version=6
  t=1.51s  stream closed by the server (order finished)

== 7. Keyset pagination
  page 1: order ids [6, 5, 4]  nextCursor=eyJiZWZvcmUiOiA0fQ
  page 2: order ids [3, 2, 1]  nextCursor=None

== 8. Transactional outbox -> signed webhooks (with one failure and a retry)
  receiver: evt_1: receiver 'down' -> 503
  receiver: evt_2: accepted order.created order 2 -> pending
  receiver: evt_3: accepted order.status_changed order 1 -> paid
  receiver: evt_4: accepted order.status_changed order 1 -> preparing
  ... 10 of 10 events accepted, each signature verified
  outbox row: (1, 'order.created', 'delivered', 1, 'HTTP 503')
  outbox row: (2, 'order.created', 'delivered', 0, None)

== 9. Rate limiting
  6 wrong passwords -> [401, 401, 401, 401, 401, 429]  Retry-After: 59 s

== 10. Errors are RFC 9457 problem documents
  422 application/problem+json
  {
    "type": "https://docs.brew.example/problems/validation-error",
    "title": "Unprocessable Content",
    "status": 422,
    "detail": "the request is invalid",
    "instance": "/v1/orders",
    "requestId": "31fb5ec2a2804788",
    "errors": [
      {
        "loc": [
          "query",
          "limit"
        ],
        "msg": "Input should be greater than or equal to 1",
        "type": "greater_than_equal"
      }
    ]
  }

== 11. Health checks
  GET /livez -> {'status': 'ok'}   GET /readyz -> {'status': 'ready', 'database': 'ok'}

== 12. The API's JSON logs: one request by its id, then the graceful shutdown
  {"level": "INFO", "logger": "orders.access", "msg": "request", "request_id": "demo-patch-0001", "method": "PATCH", "route": "/v1/orders/{order_id}", "status": 200, "user_id": 1, "duration_ms": 66.1}
  INFO  uvicorn.error  Waiting for application shutdown.
  INFO  orders.app     stopped
  INFO  uvicorn.error  Application shutdown complete.
  INFO  uvicorn.error  Finished server process [10852]

Read the output as a story. Section 3: a network retry and a five-way double-click create exactly one order. Section 4: another customer cannot even learn that order 1 exists. Section 5: a stale edit is refused. Section 6: the order moves through the kitchen live over SSE. Section 8: the partner’s endpoint was “down” for the first delivery, and the outbox retried it. Section 12: SIGTERM leads to a clean shutdown.

4. A guided tour of the code

The excerpts below are copied from the project files by the script that builds this page, so they match the code that was tested.

Lifespan

All long-lived resources are created in the lifespan and yielded as state, so tests get a fresh set per app and shutdown closes everything in order: stop taking traffic, finish the relay pass, close the HTTP client, dispose the engine.

orders_api/main.py · lifespan (inside create_app)

@asynccontextmanager
async def lifespan(app: FastAPI):
    configure_logging(settings.log_level)
    engine = make_engine(settings.database_url)
    if settings.create_tables:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    if settings.seed_demo_data:
        await seed(sessions, settings)
    http = httpx.AsyncClient(transport=webhook_transport, timeout=httpx.Timeout(5.0, connect=2.0))
    health = HealthState()
    stop = asyncio.Event()
    relay = (asyncio.create_task(relay_forever(sessions, http, settings, stop), name="outbox-relay")
             if settings.webhook_url and settings.outbox_relay_enabled else None)
    log.info("started", extra={"fields": {"database": engine.url.render_as_string(hide_password=True),
                                          "outbox_relay": relay is not None}})
    try:
        yield {"settings": settings, "sessions": sessions, "http": http, "health": health,
               "limiter": SlidingWindowLimiter(), "events": OrderEvents(),
               "dummy_hash": hash_password("timing-equaliser", settings.bcrypt_rounds)}
    finally:
        health.draining = True
        stop.set()
        if relay:
            await relay                                   # finish the current delivery pass
        await http.aclose()
        await engine.dispose()
        log.info("stopped")

Idempotency

The order, its outbox event and the idempotency record commit in one transaction. If a concurrent request with the same key commits first, the unique primary key makes this commit fail, and the loser replays the winner’s stored response.

orders_api/services.py · create_order

async def create_order(session: AsyncSession, principal: Principal, body: OrderIn,
                       idempotency_key: str) -> tuple[int, dict, bool]:
    """Returns (status, body, replayed). The order, its outbox event and the key commit together."""
    request_hash = hashlib.sha256(body.model_dump_json(by_alias=True).encode()).hexdigest()
    if replay := await _replay(session, principal, idempotency_key, request_hash):
        return *replay, True

    order = await _build_order(session, principal, body)
    session.add(order)
    await session.flush()                                   # assigns the id and defaults
    session.add(order_event("order.created", order))
    response = OrderOut.model_validate(order).model_dump(mode="json", by_alias=True)
    session.add(IdempotencyRecord(user_id=principal.id, key=idempotency_key, request_hash=request_hash,
                                  status_code=201, response_body=json.dumps(response)))
    try:
        await session.commit()
    except IntegrityError:                                  # a concurrent retry with the same key won
        await session.rollback()
        if replay := await _replay(session, principal, idempotency_key, request_hash):
            return *replay, True
        raise
    return 201, response, False

If-Match

Two layers of protection against lost updates: the If-Match check catches a client editing an old copy, and SQLAlchemy’s version_id_col adds WHERE version = … to the UPDATE, catching a concurrent writer between our read and our write.

orders_api/routers/orders.py · edit

@router.patch("/{order_id}", responses=NOT_FOUND | PRECONDITIONS | CONFLICT)
async def edit(order_id: int, patch: OrderPatch, response: Response, session: Session,
               principal: CurrentUser, if_match: IfMatch = None) -> OrderOut:
    order = await load_order(session, order_id, principal)
    check_if_match(if_match, order, required=True)
    if order.status != "pending":
        raise ApiError(409, "only pending orders can be edited", type_="invalid-state")
    for field, value in patch.model_dump(exclude_unset=True).items():     # only what was sent
        setattr(order, field, value)
    order.updated_at = utcnow()
    await _commit(session)
    response.headers["ETag"] = etag_of(order)
    return order

orders_api/services.py · check_if_match

def check_if_match(if_match: str | None, order: Order, *, required: bool) -> None:
    if if_match is None:
        if required:
            raise ApiError(428, "send If-Match with the order's current ETag", type_="precondition-required")
        return
    if not etag_matches(if_match, etag_of(order)):
        raise ApiError(412, "the order was changed by someone else; GET it again and retry",
                       type_="precondition-failed", currentEtag=etag_of(order))

SSE

The stream authorizes in a dependency, before any bytes are sent, so a stranger gets a proper 404 instead of a broken stream. It subscribes before reading the current status, so no update can fall between the two, and it uses a short session instead of the request-scoped one, because a stream may stay open for minutes.

orders_api/routers/orders.py · _subscription and events

async def _subscription(order_id: int, request: Request,
                        principal: CurrentUser) -> AsyncIterator[tuple[dict, object]]:
    """Authorize and subscribe *before* the stream starts, with a short-lived session.

    The stream may stay open for minutes; holding a request-scoped session that long would
    pin a pooled database connection per viewer.
    """
    with request.state.events.subscribe(order_id) as queue:     # subscribe first: no update can slip by
        async with request.state.sessions() as session:
            order = await load_order(session, order_id, principal)
            first = status_event(order)
        yield first, queue


@router.get("/{order_id}/events", response_class=EventSourceResponse, responses=NOT_FOUND)
async def events(subscription: Annotated[tuple[dict, object], Depends(_subscription)]
                 ) -> AsyncIterable[ServerSentEvent]:
    """Server-Sent Events: the current status now, then every change, until the order is finished."""
    event, queue = subscription
    while True:
        yield ServerSentEvent(data=event, event="status", id=str(event["version"]))
        if event["status"] in TERMINAL:
            return
        event = await queue.get()

Outbox

The relay picks due events (with SKIP LOCKED on PostgreSQL, so several replicas can run it), signs each payload, and records success, a retry with exponential backoff, or a dead letter. It holds its row locks during delivery on purpose, which is why the HTTP client has short timeouts.

orders_api/outbox.py · relay_once

async def relay_once(sessions: async_sessionmaker[AsyncSession], client: httpx.AsyncClient,
                     settings: Settings, now: Callable[[], datetime] = utcnow) -> dict[str, int]:
    counts = {"delivered": 0, "retry": 0, "dead": 0}
    async with sessions() as session:
        due = (select(OutboxEvent)
               .where(OutboxEvent.status == "pending", OutboxEvent.next_attempt_at <= now())
               .order_by(OutboxEvent.id).limit(50)
               .with_for_update(skip_locked=True))          # PostgreSQL: replicas never grab the same rows
        for event in (await session.scalars(due)).all():
            message_id, timestamp = f"evt_{event.id}", str(int(now().timestamp()))
            headers = {"content-type": "application/json", "webhook-id": message_id,
                       "webhook-timestamp": timestamp,
                       "webhook-signature": sign(settings.webhook_secret.get_secret_value(),
                                                 message_id, timestamp, event.payload)}
            try:
                response = await client.post(settings.webhook_url, content=event.payload, headers=headers)
                error = None if response.is_success else f"HTTP {response.status_code}"
            except httpx.HTTPError as exc:
                error = f"{type(exc).__name__}: {exc}"[:300]
            if error is None:
                event.status, event.delivered_at = "delivered", now()
                counts["delivered"] += 1
                continue
            event.attempts += 1
            event.last_error = error
            if event.attempts >= settings.outbox_max_attempts:
                event.status = "dead"                       # alert on these; replay after fixing the receiver
                counts["dead"] += 1
                log.error("webhook dead", extra={"fields": {"event_id": event.id, "error": error}})
            else:
                event.next_attempt_at = now() + backoff(event.attempts)
                counts["retry"] += 1
                log.warning("webhook failed, will retry",
                            extra={"fields": {"event_id": event.id, "attempt": event.attempts, "error": error}})
        await session.commit()
    return counts

Auth

Tokens are decoded with a pinned algorithm and required claims. Login spends the same bcrypt time for unknown accounts, and only failed attempts count toward the limit.

orders_api/security.py · decode_access_token and current_user

def decode_access_token(token: str, settings: Settings) -> Principal:
    claims = jwt.decode(token, settings.jwt_secret.get_secret_value(), algorithms=[ALGORITHM],
                        audience=settings.jwt_audience, issuer=settings.jwt_issuer,
                        options={"require": ["exp", "iat", "sub", "aud", "iss"]})
    return Principal(id=int(claims["sub"]), email=claims["email"], role=claims["role"])


async def current_user(request: Request, token: Annotated[str | None, Depends(oauth2_scheme)]) -> Principal:
    if token is None:
        raise ApiError(401, "missing bearer token", type_="unauthenticated", headers=BEARER)
    try:
        principal = decode_access_token(token, request.state.settings)
    except jwt.PyJWTError:
        raise ApiError(401, "invalid or expired token", type_="unauthenticated", headers=BEARER) from None
    request.state.principal = principal               # for the access log and the rate limiter
    return principal

orders_api/routers/auth.py · login

@router.post("/auth/token", responses={401: {"model": Problem}, 429: {"model": Problem}})
async def login(request: Request, form: Annotated[OAuth2PasswordRequestForm, Depends()],
                session: Annotated[AsyncSession, Depends(get_session)]) -> TokenOut:
    settings = request.state.settings
    limiter, limit = request.state.limiter, settings.login_attempts_per_minute
    key = f"login:{request.client.host if request.client else 'unknown'}:{form.username.lower()}"
    raise_if_limited(limiter.retry_after(key, limit))           # only failed attempts count
    user = await session.scalar(select(User).where(User.email == form.username.lower()))
    # unknown e-mails pay the same bcrypt cost: response times don't reveal which accounts exist
    valid = await verify_password(form.password, user.password_hash if user else request.state.dummy_hash)
    if user is None or not valid:
        limiter.hit(key, limit)
        raise ApiError(401, "incorrect e-mail or password", type_="unauthenticated")
    token, expires_in = create_access_token(Principal(user.id, user.email, user.role), settings)
    return TokenOut(access_token=token, expires_in=expires_in)

5. The tests

python -m pytest (real)

...................................................                      [100%]
51 passed in 11.74s

6. Design decisions and their trade-offs

Why is the Idempotency-Key required instead of optional?

Mobile networks drop responses. A client that retries a POST without a key can create two orders and charge twice. Making the key required turns a rare, expensive bug into an immediate, obvious 400 during development. Stripe-style APIs make it optional for backwards compatibility; a new API can afford to be strict.

Why 404 instead of 403 for other customers' orders?

A 403 confirms that order 1043 exists, which leaks business volume and helps enumeration. Scoping the query by owner makes "not yours" and "doesn't exist" the same code path, so it cannot be forgotten on a new endpoint. Admin-only operations still return 403, because the endpoint's existence is public anyway.

Why stateless JWTs, and what does that cost?

No database lookup per request, and any replica can verify a token. The cost: a demoted admin keeps their role until the token expires, hence the 15-minute lifetime. If you need instant revocation, add a denylist keyed by jti or move to refresh-token rotation with server-side sessions.

What breaks with several replicas, and what doesn't?

Works as is: the database, idempotency, optimistic locking and the outbox (with SKIP LOCKED). Per process: the rate limiter and the SSE broker. Move the limiter to Redis and publish status events through Redis pub/sub or PostgreSQL LISTEN/NOTIFY. The interfaces (SlidingWindowLimiter.hit, OrderEvents.publish) were kept small for exactly that swap.

Why create tables at start-up if production should use migrations?

So the project runs with zero setup. ORDERS_CREATE_TABLES=false turns it off; in production, run Alembic as a separate job before the rollout, with expand/contract migrations so old and new pods both work during the deploy (part 23).

7. Take it further

Recap