FastAPI Framework: A Developer-Friendly, End-to-End Guide

FastAPI is a modern, high-performance web framework for building APIs with Python 3.8+ based on standard type hints. It’s built on Starlette for the web parts and Pydantic for data validation, giving you speed, reliability, and automatic OpenAPI docs with minimal code.

This guide walks you from “hello world” to a production-ready project: models, validation, dependency injection, security with JWT, async DB access, testing, and deployment. Copy-paste friendly code is included throughout.

---

Why FastAPI?

• Performance: Among the fastest Python frameworks thanks to ASGI/uvicorn and Starlette.
• Type-Driven: Request and response models use type hints; you get validation and editor auto-completion “for free”.
• Batteries Included: Automatic OpenAPI (Swagger UI & ReDoc), dependency injection, background tasks, WebSockets.
• Async-First: First-class async support; works great with async ORMs/clients.
• Developer Experience: Minimal boilerplate, clear error messages, excellent docs.

---

Prerequisites

• Python 3.8+
• Basic familiarity with HTTP and JSON
• (Optional) Docker if you plan to containerize

---

Setup & Hello World

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install "fastapi[standard]"  # includes uvicorn + extras

Create main.py:

from fastapi import FastAPI

app = FastAPI(title="Hello FastAPI")

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

Run:

fastapi dev main.py  # auto-reload; or: uvicorn main:app --reload

Open:

• Interactive docs (Swagger UI): http://127.0.0.1:8000/docs
• ReDoc: http://127.0.0.1:8000/redoc

---

Path, Query, and Body Parameters (with Validation)

from typing import Optional, List
from fastapi import FastAPI, Path, Query, Body
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    price: float = Field(gt=0)
    tags: List[str] = []
    in_stock: bool = True

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(..., ge=1),
    detailed: bool = Query(False),
):
    return {"item_id": item_id, "detailed": detailed}

@app.post("/items", status_code=201)
def create_item(item: Item = Body(...)):
    return item

What you get automatically

• 422 responses if validation fails
• OpenAPI schema generation with your field constraints (min_length, gt, etc.)

---

Response Models & Data Shaping

Use response_model to control what you return (and to document it).

from pydantic import BaseModel

class ItemOut(BaseModel):
    name: str
    price: float
    in_stock: bool

@app.post("/items", response_model=ItemOut, status_code=201)
def create_item(item: Item):
    # save to DB here...
    return item

You can also exclude fields dynamically:

@app.get("/items/{item_id}", response_model=ItemOut, response_model_exclude={"in_stock"})
def get_item(item_id: int):
    ...

---

Dependency Injection (Core to FastAPI)

Dependencies let you share logic (DB sessions, auth, settings) cleanly.

from fastapi import Depends, Header, HTTPException

def get_token(authorization: str = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing token")
    return authorization[7:]

@app.get("/secure")
def secure_endpoint(token: str = Depends(get_token)):
    return {"ok": True}

You can scope dependencies per-router or globally (see “Routers & Structure”).

---

Background Tasks

from fastapi import BackgroundTasks

def send_email(to: str, subject: str, body: str):
    # call SMTP provider, etc.
    pass

@app.post("/contact")
def contact(email: str, background: BackgroundTasks):
    background.add_task(send_email, to=email, subject="Thanks", body="We got your message.")
    return {"status": "queued"}

---

Middleware & CORS

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.app"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Custom middleware example:

from starlette.middleware.base import BaseHTTPMiddleware
import time, logging

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = (time.perf_counter() - start) * 1000
        logging.info("%s %s -> %sms", request.method, request.url.path, round(duration, 2))
        return response

app.add_middleware(TimingMiddleware)

---

Handling Errors

from fastapi import HTTPException
from fastapi.responses import JSONResponse
from fastapi.requests import Request

@app.exception_handler(KeyError)
async def key_error_handler(request: Request, exc: KeyError):
    return JSONResponse(status_code=400, content={"detail": f"Key missing: {exc}"})

@app.get("/boom")
def boom():
    raise HTTPException(status_code=404, detail="Not found")

---

Security: OAuth2 Password Flow + JWT

Install crypto:

pip install python-jose[cryptography] passlib[bcrypt]

Auth scaffolding (auth.py):

from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status, APIRouter
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "change-me"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

fake_users = {
    "alice": pwd_context.hash("wonderland"),
}

class Token(BaseModel):
    access_token: str
    token_type: str

router = APIRouter(prefix="/auth", tags=["auth"])

def authenticate(username: str, password: str) -> bool:
    hashed = fake_users.get(username)
    return hashed and pwd_context.verify(password, hashed)

def create_access_token(sub: str, expires_delta: Optional[timedelta] = None):
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    return jwt.encode({"sub": sub, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

@router.post("/token", response_model=Token)
def login(form: OAuth2PasswordRequestForm = Depends()):
    if not authenticate(form.username, form.password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    token = create_access_token(sub=form.username)
    return {"access_token": token, "token_type": "bearer"}

def require_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload["sub"]
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Use it in routes:

from fastapi import Depends
from .auth import router as auth_router, require_user

app.include_router(auth_router)

@app.get("/me")
def me(user: str = Depends(require_user)):
    return {"user": user}

---

Project Structure with Routers & Settings

A clean, scalable layout:

app/
  __init__.py
  main.py
  dependencies.py
  api/
    __init__.py
    v1/
      __init__.py
      items.py
      users.py
  models/
    __init__.py
    item.py
    user.py
  db/
    __init__.py
    session.py
  core/
    config.py
    security.py
tests/
  test_items.py

core/config.py with pydantic-settings

pip install pydantic-settings

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    debug: bool = False
    database_url: str = "sqlite+aiosqlite:///./app.db"
    secret_key: str = "change-me"

    class Config:
        env_file = ".env"

settings = Settings()

Use in main.py:

from fastapi import FastAPI
from .core.config import settings

app = FastAPI(title=settings.app_name, debug=settings.debug)

---

Async Database Access (SQLAlchemy 2.0 + Async)

Install:

pip install "sqlalchemy[asyncio]" aiosqlite

db/session.py:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from contextlib import asynccontextmanager
from ..core.config import settings

engine = create_async_engine(settings.database_url, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@asynccontextmanager
async def get_session() -> AsyncSession:
    async with SessionLocal() as session:
        yield session

models/item.py:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String, Float, Boolean

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)
    price: Mapped[float] = mapped_column(Float)
    in_stock: Mapped[bool] = mapped_column(Boolean, default=True)

Create tables on startup:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncEngine
from .db.session import engine
from .models.item import Base

app = FastAPI()

@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

CRUD route (api/v1/items.py):

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from ...db.session import get_session
from ...models.item import Item

router = APIRouter(prefix="/items", tags=["items"])

class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    price: float = Field(gt=0)
    in_stock: bool = True

class ItemOut(ItemIn):
    id: int

@router.post("", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
async def create_item(payload: ItemIn, db: AsyncSession = Depends(get_session)):
    item = Item(**payload.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

@router.get("/{item_id}", response_model=ItemOut)
async def read_item(item_id: int, db: AsyncSession = Depends(get_session)):
    res = await db.execute(select(Item).where(Item.id == item_id))
    item = res.scalar_one_or_none()
    if not item:
        raise HTTPException(404, "Item not found")
    return item

Include router in main.py:

from .api.v1.items import router as items_router
app.include_router(items_router, prefix="/api/v1")

---

WebSockets & Streaming (Quick Peek)

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/echo")
async def ws_echo(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"echo: {data}")
    except WebSocketDisconnect:
        pass

Server-sent events / streaming responses use StreamingResponse from Starlette.

---

Pagination & Filtering

Simple page/limit:

from typing import List
from sqlalchemy import select

@app.get("/items", response_model=List[ItemOut])
async def list_items(page: int = 1, limit: int = 20, db: AsyncSession = Depends(get_session)):
    offset = (page - 1) * limit
    res = await db.execute(select(Item).offset(offset).limit(limit))
    return list(res.scalars())

---

Rate Limiting (Lightweight Example)

There’s no built-in rate limiter; use middleware or a proxy (e.g., Nginx). For demo purposes only, an in-memory token bucket:

import time
from fastapi import Request, HTTPException

CALLS = {}

@app.middleware("http")
async def simple_rate_limit(request: Request, call_next):
    key = request.client.host
    window = 60
    limit = 100
    now = int(time.time())

    bucket = CALLS.get(key, {"ts": now, "count": 0})
    if now - bucket["ts"] >= window:
        bucket = {"ts": now, "count": 0}

    if bucket["count"] >= limit:
        raise HTTPException(429, "Too Many Requests")

    bucket["count"] += 1
    CALLS[key] = bucket
    return await call_next(request)

(For production, use Redis-backed limiters or API gateways.)

---

Testing with pytest & TestClient

pip install pytest httpx

tests/test_items.py:

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_root():
    r = client.get("/")
    assert r.status_code == 200
    assert r.json()["message"] == "Hello, FastAPI!"

Run: pytest -q

For async routes, TestClient handles the loop; for more control, use httpx.AsyncClient with lifespan="on".

---

Logging & Observability

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s :: %(message)s",
)

logger = logging.getLogger("app")
logger.info("App starting...")

Add request IDs (via custom middleware) and export logs to stdout for container platforms. Consider OpenTelemetry for tracing.

---

Configuration by Environment

• Keep secrets in .env or environment variables.
• Provide sensible defaults in Settings.
• In Docker, pass ENV vars at runtime.

---

Dockerization (Production-Ready)

Dockerfile:

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

WORKDIR /app

RUN apt-get update && apt-get install -y build-essential && rm -rf /var/lib/apt/lists/*

COPY pyproject.toml poetry.lock* /app/
RUN pip install --upgrade pip && pip install "fastapi[standard]" "sqlalchemy[asyncio]" aiosqlite pydantic-settings python-jose[cryptography] passlib[bcrypt]

COPY . /app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

(Use gunicorn with uvicorn.workers.UvicornWorker on larger instances.)

---

Performance Tips

• Prefer async DB drivers & HTTP clients to avoid blocking the loop.
• Use response_model to serialize only what you need.
• Tune workers (--workers) and keep worker count ≈ CPU cores.
• Enable HTTP keep-alive / connection pooling in clients.
• Profile endpoints (e.g., py-spy, scalene) to find bottlenecks.

---

Common Gotchas

• Mixing sync & async: Blocking I/O in async def routes will stall the loop. Use async clients or offload with run_in_executor.
• Validation surprises: Pydantic casts types (e.g., "123" to int). Use stricter validators if you need hard errors.
• Circular imports: Keep routers, models, and settings modular; avoid importing app at module import time.

---

A Mini “Real” Example: Todo API with JWT & Async DB

app/main.py:

from fastapi import FastAPI
from .core.config import settings
from .api.v1.todos import router as todos_router
from .auth import router as auth_router
from .db.session import engine
from .models.todo import Base

app = FastAPI(title=settings.app_name)

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

app.include_router(auth_router, prefix="/api/v1")
app.include_router(todos_router, prefix="/api/v1")

models/todo.py:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String, Boolean

class Base(DeclarativeBase): pass

class Todo(Base):
    __tablename__ = "todos"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(120), index=True)
    done: Mapped[bool] = mapped_column(Boolean, default=False)
    owner: Mapped[str] = mapped_column(String(50), index=True)

api/v1/todos.py:

from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from ...db.session import get_session
from ...models.todo import Todo
from ...auth import require_user

router = APIRouter(prefix="/todos", tags=["todos"])

class TodoIn(BaseModel):
    title: str = Field(min_length=1, max_length=120)
    done: bool = False

class TodoOut(TodoIn):
    id: int

@router.post("", response_model=TodoOut, status_code=status.HTTP_201_CREATED)
async def create_todo(
    payload: TodoIn,
    user: str = Depends(require_user),
    db: AsyncSession = Depends(get_session),
):
    todo = Todo(**payload.model_dump(), owner=user)
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return todo

@router.get("", response_model=List[TodoOut])
async def list_todos(user: str = Depends(require_user), db: AsyncSession = Depends(get_session)):
    res = await db.execute(select(Todo).where(Todo.owner == user))
    return list(res.scalars())

@router.put("/{todo_id}", response_model=TodoOut)
async def update_todo(
    todo_id: int,
    payload: TodoIn,
    user: str = Depends(require_user),
    db: AsyncSession = Depends(get_session),
):
    res = await db.execute(select(Todo).where(Todo.id == todo_id, Todo.owner == user))
    todo = res.scalar_one_or_none()
    if not todo:
        raise HTTPException(404, "Not found")

    await db.execute(
        update(Todo)
        .where(Todo.id == todo_id)
        .values(**payload.model_dump())
    )
    await db.commit()
    await db.refresh(todo)
    return todo

@router.delete("/{todo_id}", status_code=204)
async def delete_todo(
    todo_id: int,
    user: str = Depends(require_user),
    db: AsyncSession = Depends(get_session),
):
    res = await db.execute(select(Todo).where(Todo.id == todo_id, Todo.owner == user))
    if not res.scalar_one_or_none():
        raise HTTPException(404, "Not found")
    await db.execute(delete(Todo).where(Todo.id == todo_id))
    await db.commit()

Test flow:

1. POST /api/v1/auth/token with form username=alice&password=wonderland.
2. Use returned access_token as Authorization: Bearer <token>.
3. CRUD on /api/v1/todos.

---

Documentation & Versioning

• FastAPI generates OpenAPI schema at /openapi.json.
• Multiple versions: mount routers under /api/v1, /api/v2. Maintain backward compatibility by deprecating endpoints gradually.

---

Deploy Checklist

• Set strong SECRET_KEY; rotate regularly.
• Configure CORS for your domain(s).
• Use HTTPS (proxy/ingress termination).
• Run multiple workers; use a process manager (systemd, Docker, Kubernetes).
• Structured JSON logs; aggregate in CloudWatch/ELK.
• Health checks (/healthz) and readiness checks.
• Database migrations (e.g., Alembic) for schema changes.
• Observability (metrics/tracing) and error reporting (Sentry, etc.).
• Backups for persistent data.

---

FAQ

Is FastAPI only for APIs?
Primarily yes, but you can serve server-rendered templates via Starlette if needed. Most pair FastAPI with a frontend SPA or mobile clients.

Sync or async routes?
Use async def for IO-bound operations with async clients/drivers. Pure CPU-bound tasks should run in a worker (Celery/RQ) or process pool.

Which ORM?
SQLAlchemy (sync or async) is the most common. Alternatives: Tortoise ORM, Gino, or SQLModel (a Pydantic-flavored wrapper over SQLAlchemy).

---

Conclusion

FastAPI’s combination of speed, type-safety, and developer experience makes it an excellent choice for modern backends. Start small with a single main.py, then evolve into routers, dependencies, and async DB access. With good structure, testing, and deployment hygiene, you’ll have a production-grade API that’s a joy to maintain.

If you’d like, I can turn this into a ready-to-run template repo (with Docker, Alembic, CI, and a sample frontend) or tailor the examples to your exact use case.