FastAPI Cheatsheet: Routing, Pydantic, Auth and Testing

# Install
pip install "fastapi[standard]"
# ↑ bundles uvicorn, httpx, jinja2, python-multipart, email-validator

# Run (dev — auto-reload, docs at /docs)
fastapi dev app.py

# Run (prod)
fastapi run app.py --workers 4

# Or directly with uvicorn
uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

# Alembic migrations (if using SQLAlchemy)
pip install alembic sqlalchemy "psycopg[binary]"

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Items API", version="1.0")

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

@app.get("/")
async def root() -> dict:
    return {"status": "ok"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

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

# fastapi dev app.py  →  http://127.0.0.1:8000/docs

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

# Plain dependency
async def common_params(q: str = "", skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

CommonDeps = Annotated[dict, Depends(common_params)]

@app.get("/items")
async def list_items(params: CommonDeps):
    return params

# Auth via dependency
async def require_token(x_token: Annotated[str, Header()]):
    if x_token != "secret":
        raise HTTPException(401, "bad token")
    return {"user": "alice"}

@app.get("/me")
async def me(user: Annotated[dict, Depends(require_token)]):
    return user

# Class-based dependency
class Pager:
    def __init__(self, page: int = 1, size: int = 20):
        self.offset = (page - 1) * size
        self.limit = size

@app.get("/posts")
async def posts(p: Annotated[Pager, Depends()]):
    return {"offset": p.offset, "limit": p.limit}

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.http = httpx.AsyncClient(timeout=10)
    yield
    # Shutdown
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/proxy")
async def proxy():
    r = await app.state.http.get("https://httpbin.org/get")
    return r.json()

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext

SECRET, ALG = "change-me", "HS256"
pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2 = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()
USERS = {"alice": {"hashed": pwd.hash("wonderland")}}

def make_token(sub: str) -> str:
    payload = {"sub": sub,
               "exp": datetime.now(timezone.utc) + timedelta(minutes=30)}
    return jwt.encode(payload, SECRET, algorithm=ALG)

@app.post("/token")
async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    u = USERS.get(form.username)
    if not u or not pwd.verify(form.password, u["hashed"]):
        raise HTTPException(401, "bad credentials")
    return {"access_token": make_token(form.username), "token_type": "bearer"}

async def current_user(token: Annotated[str, Depends(oauth2)]) -> str:
    try:
        return jwt.decode(token, SECRET, algorithms=[ALG])["sub"]
    except JWTError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")

@app.get("/me")
async def me(user: Annotated[str, Depends(current_user)]):
    return {"user": user}

# tests/test_app.py
import pytest
from httpx import AsyncClient, ASGITransport
from fastapi.testclient import TestClient
from app import app

# 1 · Sync — quick smoke
def test_root():
    client = TestClient(app)
    r = client.get("/")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}

# 2 · Async — real concurrency
@pytest.mark.asyncio
async def test_item():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://t"
    ) as c:
        r = await c.post("/items", json={"name": "tea", "price": 3.5})
        assert r.status_code == 201
        assert r.json()["name"] == "tea"

# 3 · Override a dependency
from app import require_token
def fake_token():
    return {"user": "tester"}
app.dependency_overrides[require_token] = fake_token

from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DB_URL = "sqlite+aiosqlite:///./app.db"
engine = create_async_engine(DB_URL)
Session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase): pass
class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str] = mapped_column(index=True)

class NoteIn(BaseModel):
    text: str = Field(min_length=1, max_length=200)

class NoteOut(NoteIn):
    id: int

@asynccontextmanager
async def lifespan(app):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

app = FastAPI(lifespan=lifespan)

async def db() -> AsyncSession:
    async with Session() as s:
        yield s

DB = Annotated[AsyncSession, Depends(db)]

@app.post("/notes", response_model=NoteOut, status_code=201)
async def create(payload: NoteIn, s: DB):
    n = Note(text=payload.text)
    s.add(n); await s.commit(); await s.refresh(n)
    return n

@app.get("/notes/{nid}", response_model=NoteOut)
async def get(nid: int, s: DB):
    n = await s.get(Note, nid)
    if not n: raise HTTPException(404, "not found")
    return n