Learn / Python and FastAPI Backend Engineering
Deploys, Workers, and Runbooks
Finish the FastAPI service with graceful worker shutdown, traffic-safe readiness gates, deploy smoke checks, rollback thresholds, and a two-minute handoff runbook.
Course: Python and FastAPI Backend Engineering. Level: Intermediate. Topic: Backend craft.
Stage: pro - Pro backend operations - Deploys, workers, and FastAPI runbooks. Finish the FastAPI path with background work, graceful shutdown, deploy smoke checks, and rollback-ready operations.
Outcomes
- Use FastAPI lifespan to start background workers, mark readiness, and shut down without dropping or duplicating work.
- Write deploy smoke checks that prove HTTP contracts, readiness, logs, and workers survived the release.
- Define rollback thresholds and a runbook another engineer can execute during an incident.
- Plan a small FastAPI service from requirements.
- Implement routes, models, services, repositories, and SQLite persistence.
- Add tests for core services and HTTP behavior.
- Document local setup, quality checks, and architecture decisions.
Concepts
- lifespan orchestration
- cancellable worker
- deploy smoke check
- rollback threshold
- handoff runbook
- Final Project
- Project brief
- Suggested structure
- Python + FastAPI foundations
- Guided practice
Concept flow
Show how startup, readiness, worker shutdown, smoke checks, rollback thresholds, and handoff notes connect into one release packet.
- Deploy starts
- Lifespan startup
- Readiness gate
- Worker loop
- Smoke checks
- Rollback decision
- Handoff runbook
Session flow
- Use lifespan as the operations boundary (concept, 12 min) — Start dependencies, workers, and readiness state from one explicit FastAPI lifespan boundary.
- Startup is not complete until required dependencies have been checked.
- Readiness should be set intentionally, not inferred from process existence.
- Make workers deploy-safe (walkthrough, 18 min) — Build a background worker that observes shutdown, records lag, and relies on idempotent job handling.
- A worker must stop claiming new work when shutdown begins.
- Idempotency and retry rules are part of deploy safety, not only job logic.
- Turn deploy smoke into evidence (exercise, 14 min) — Write exact post-deploy commands for liveness, readiness, one contract route, logs, metrics, and worker lag.
- Smoke checks need expected output and owner action on failure.
- Checking liveness alone can hide unsafe dependencies and broken background work.
- Write the two-minute runbook (exercise, 14 min) — Document symptoms, first queries, rollback thresholds, mitigation commands, escalation owner, and customer-safe wording.
- A runbook should help someone who did not build the service.
- Rollback is faster when the threshold and command are written before the incident.
- Capstone: Task Tracker API: Project brief (concept, 150 min) — Build an API for tasks with title, optional description, status, priority, and timestamps. The API should support create, list with filters, retrieve, update status, and delete. It should use SQLite locally and expose OpenAPI docs through FastAPI.
- Retained source example: Required endpoints
POST /tasks
GET /tasks?status=open&limit=20
GET /tasks/{task_id}
PATCH /tasks/{task_id}/status
DELETE /tasks/{task_id}
GET /health
- Capstone: Task Tracker API: Suggested structure (walkthrough, 150 min) — Keep the app small but layered. Use routes for HTTP, schemas for API models, services for workflows, repositories for SQL, and tests for both service behavior and route contracts.
- Retained source example: Capstone layout
app/
main.py
routers/tasks.py
schemas.py
services/tasks.py
repositories/tasks_sqlite.py
database.py
tests/
test_task_service.py
test_task_api.py
- Capstone: Task Tracker API: practices and mistakes (review, 5 min) — Retained guidance from the legacy PTesting lesson.
- Practice: Define acceptance criteria before coding.
- Practice: Use dependency overrides so API tests can use an isolated database.
- Practice: Document trade-offs, especially around SQLite and sync versus async choices.
- Avoid: Starting with database tables before clarifying API behavior.
- Avoid: Skipping tests for validation and not-found cases.
- Avoid: Letting the final project become too large to finish and review.
- Capstone: Task Tracker API: references (review, 2 min) — Original references retained from the legacy library.
- FastAPI documentation: https://fastapi.tiangolo.com/
- Pydantic documentation: https://docs.pydantic.dev/
- FastAPI SQL databases: https://fastapi.tiangolo.com/tutorial/sql-databases/
- pytest documentation: https://docs.pytest.org/
Code example
Python in app/lifespan.py.
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass
from fastapi import FastAPI, HTTPException, status
@dataclass
class RuntimeState:
ready: bool = False
worker_lag_seconds: int = 0
last_worker_error: str | None = None
async def run_outbox_worker(stop_event: asyncio.Event, state: RuntimeState) -> None:
while not stop_event.is_set():
try:
job = await outbox.claim_next(timeout=2)
if job is None:
continue
await publish_event(job)
await outbox.mark_published(job.id)
state.worker_lag_seconds = await outbox.lag_seconds()
state.last_worker_error = None
except Exception as exc:
state.last_worker_error = exc.__class__.__name__
logger.exception("outbox_worker_failed")
await asyncio.sleep(1)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
state = RuntimeState()
stop_event = asyncio.Event()
worker: asyncio.Task[None] | None = None
app.state.runtime = state
try:
await database.ping()
state.ready = True
worker = asyncio.create_task(run_outbox_worker(stop_event, state))
yield
finally:
state.ready = False
stop_event.set()
if worker is not None:
try:
await asyncio.wait_for(worker, timeout=10)
except TimeoutError:
worker.cancel()
with suppress(asyncio.CancelledError):
await worker
app = FastAPI(lifespan=lifespan)
@app.get("/health/live", include_in_schema=False)
async def live() -> dict[str, str]:
return {"status": "alive"}
@app.get("/health/ready", include_in_schema=False)
async def ready() -> dict[str, str | int]:
state: RuntimeState = app.state.runtime
if not state.ready:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Service is not ready.")
if state.worker_lag_seconds > 120:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Worker lag is above threshold.")
return {"status": "ready", "worker_lag_seconds": state.worker_lag_seconds}
Walkthrough examples
- FastAPI deploy operations packet — A backend team is shipping a billing event worker with the API. The release must prove traffic safety, worker drain behavior, and rollback readiness.
- File: app/main.py
- File: app/lifespan.py
- File: app/workers/outbox.py
- File: tests/test_lifespan_readiness.py
- File: tests/test_worker_shutdown.py
- File: docs/ops/deploy-runbook.md
- Add RuntimeState and lifespan so startup checks dependencies before setting readiness.
- Start the outbox worker from lifespan and stop it with a bounded cancellation path.
- Expose /health/live and /health/ready with worker lag and dependency thresholds.
- Write tests for startup readiness, shutdown drain, worker retry/idempotency, and readiness failure on lag.
- Write deploy smoke commands and rollback thresholds in docs/ops/deploy-runbook.md.
- Required endpoints — Additional example retained from the legacy lesson.
- Retained source code:
POST /tasks
GET /tasks?status=open&limit=20
GET /tasks/{task_id}
PATCH /tasks/{task_id}/status
DELETE /tasks/{task_id}
GET /health
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Capstone layout — Additional example retained from the legacy lesson.
- Retained source code:
app/
main.py
routers/tasks.py
schemas.py
services/tasks.py
repositories/tasks_sqlite.py
database.py
tests/
test_task_service.py
test_task_api.py
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
- Status model — Additional example retained from the legacy lesson.
- Retained source code:
from typing import Literal
from pydantic import BaseModel, Field
TaskStatus = Literal['open', 'doing', 'done']
class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=120)
description: str | None = Field(default=None, max_length=1000)
- Explain the expected behavior.
- Compare the example with the canonical PTLearn implementation.
Practice
- Add a lifespan-managed worker that starts after dependency checks and exits when the app shuts down.
- Make readiness fail when startup is incomplete, a required dependency is unavailable, or worker lag crosses the threshold.
- Write exact deploy smoke commands for live, ready, one contract route, one log query, and one worker signal.
- Define rollback thresholds for error rate, readiness failures, worker lag, and failed smoke checks.
- Write a two-minute runbook with symptoms, first commands, owner escalation, rollback command, and customer-safe update.
- Acceptance checklist: Select the acceptance criterion that best proves the API validates bad input.
- Hint: Think about FastAPI validation responses.
- Reference solution: A test posts an empty title and expects HTTP 422.
- Accepted answers: A test posts an empty title and expects HTTP 422.
Checklist
- Lifespan startup pings required dependencies before setting ready=true.
- Background workers observe cancellation and finish without duplicate publishes.
- Readiness checks traffic safety, not just process liveness.
- Deploy smoke commands include expected output and failure interpretation.
- Rollback thresholds are numeric, observable, and owned by the deployer.
- The handoff runbook starts from symptoms and ends with mitigate, rollback, or escalate.
Quiz prompts
- Why should readiness turn false before shutting down workers? — Readiness is a traffic gate. During shutdown, stop new traffic first, then let in-flight work and workers exit safely.
- Which worker behavior is safest during deploy shutdown? — Deploy-safe workers need bounded shutdown and idempotent retry semantics so deploys do not lose or duplicate work.
- What makes a deploy smoke check reviewable? — A smoke check is operational evidence when another engineer can run it and know what result should trigger rollback.
- Which signal is a strong rollback threshold for this lesson? — Rollback thresholds should be observable, numeric, and tied to user or operations risk.
- A handoff runbook is useful only if it starts from what? — Incident runbooks should begin where the next owner actually begins: an alert, a symptom, and a safe first action.
- Which artifact best explains how to run and verify the final project? — The README should document setup, run commands, tests, and design notes.
Flashcards
- What does FastAPI lifespan own in this lesson? Startup checks, runtime readiness state, background worker startup, and graceful shutdown.
- How is readiness different from liveness during deploy? Liveness says the process can respond. Readiness says this instance should receive user traffic now.
- What makes a background worker deploy-safe? It observes shutdown, stops claiming new work, handles retries idempotently, and exits within a bounded time.
- What belongs in a deploy smoke check? Exact command, expected output, failure meaning, owner action, and rollback threshold.
- What makes a rollback threshold useful? It is numeric, observable, tied to user or operations risk, and written before the deploy starts.
- What should a two-minute handoff runbook optimize for? A next owner can start from a symptom, run the first safe query, decide mitigate/rollback/escalate, and update users safely.
Labs
- Ship a deploy-safe FastAPI operations packet — Finish the FastAPI service with lifespan startup, worker shutdown proof, readiness gates, deploy smoke checks, rollback thresholds, and a handoff runbook.
- Move startup checks, readiness state, and worker startup into FastAPI lifespan.
- Make the worker observe a stop event or cancellation and prove it exits within a bounded timeout.
- Add readiness failure cases for dependency outage and worker lag above threshold.
- Write post-deploy smoke commands for live, ready, one API contract, logs, metrics, and worker lag.
- Document rollback thresholds and a two-minute runbook another engineer can execute.
- Tests prove readiness is false before startup completes and during shutdown.
- Worker shutdown evidence shows no duplicate publish and a bounded exit path.
- Smoke checks include exact commands, expected output, and owner action on failure.
- Rollback thresholds include error rate, readiness failure rate, worker lag, and failed smoke check count.
- The runbook starts from symptoms and ends with mitigation, rollback, or escalation.
Challenge
- Pro FastAPI deploy and handoff packet (Capstone) — Create the final operations packet for a production FastAPI service: lifespan startup, cancellable worker, readiness gate, smoke checks, rollback thresholds, and handoff runbook.
- Lifespan owns startup checks, readiness state, worker startup, and graceful shutdown.
- Worker tests prove cancellation/drain behavior, idempotent retry, and bounded exit.
- Readiness fails for dependency outage, startup incomplete, shutdown, and worker lag above threshold.
- Deploy smoke commands include expected output and clear rollback criteria.
- The handoff runbook names first query, dashboards, rollback command, escalation owner, and customer-safe update.
Canonical lesson URL