FastAPI

uv init api && cd api
uv add "fastapi[standard]"   # CLI, uvicorn, httpx, ...
uv run fastapi dev           # reload, 127.0.0.1:8000
uv run fastapi run           # prod, 0.0.0.0:8000
# pip: python -m venv .venv
#      pip install "fastapi[standard]"

from fastapi import FastAPI
 
app = FastAPI(title="Notes API", version="1.0.0")
 
 
@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}

from enum import StrEnum
from typing import Annotated
from uuid import UUID
 
from fastapi import (
    Body, Cookie, FastAPI, Header, Path, Query,
)
from pydantic import BaseModel, Field
 
app = FastAPI()
 
 
class Sort(StrEnum):
    new = "new"
    top = "top"
 
 
@app.get("/items/{item_id}")
async def get_item(
    item_id: Annotated[int, Path(ge=1)],
    q: Annotated[str | None, Query(max_length=50)] = None,
    sort: Sort = Sort.new,                    # ?sort=top
    tags: Annotated[list[str], Query()] = [],  # ?tags=a
    x_request_id: Annotated[UUID | None, Header()] = None,
    session: Annotated[str | None, Cookie()] = None,
) -> dict[str, object]:
    return {"id": item_id, "q": q, "sort": sort}
 
 
class ItemIn(BaseModel):
    name: str
    price: float = Field(gt=0)
 
 
@app.put("/items/{item_id}")
async def put_item(
    item_id: int,                           # path
    item: ItemIn,                           # JSON body
    note: Annotated[str, Body()] = "",      # extra body key
) -> dict[str, object]:
    return {"id": item_id, **item.model_dump(), "note": note}
 
 
@app.get("/files/{path:path}")              # /files/a/b.txt
async def read_file(path: str) -> dict[str, str]:
    return {"path": path}

from datetime import datetime
from typing import Annotated, Self
 
from pydantic import (
    AfterValidator, BaseModel, ConfigDict, EmailStr, Field,
    StringConstraints, computed_field, field_validator,
    model_validator,
)
 
Slug = Annotated[str, StringConstraints(
    pattern=r"^[a-z0-9-]+$", max_length=40,
)]
Tag = Annotated[str, AfterValidator(str.casefold)]
 
 
class UserBase(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        extra="forbid",               # 422 on unknown keys
    )
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)
    handle: Slug
    tags: list[Tag] = []
 
 
class UserCreate(UserBase):
    password: str = Field(min_length=12, repr=False)
    confirm: str
 
    @field_validator("name")
    @classmethod
    def not_reserved(cls, v: str) -> str:
        if v.lower() == "admin":
            raise ValueError("reserved name")
        return v
 
    @model_validator(mode="after")
    def passwords_match(self) -> Self:
        if self.password != self.confirm:
            raise ValueError("passwords differ")
        return self
 
 
class UserOut(UserBase):
    model_config = ConfigDict(from_attributes=True)  # ORM
    id: int
    created_at: datetime
 
    @computed_field  # type: ignore[prop-decorator]
    @property
    def url(self) -> str:
        return f"/users/{self.id}"

@app.post("/users", status_code=201, response_model=UserOut)
async def create_user(body: UserCreate) -> dict[str, object]:
    data = body.model_dump(exclude={"password", "confirm"})
    return {**data, "id": 1, "created_at": datetime.now()}
 
 
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> UserOut:  # preferred
    ...

from fastapi import HTTPException, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
 
 
@app.get("/things/{thing_id}")
async def get_thing(thing_id: int) -> dict[str, int]:
    if thing_id == 0:
        raise HTTPException(
            status.HTTP_404_NOT_FOUND,
            detail="Thing not found",       # any JSON value
            headers={"X-Reason": "missing"},
        )
    return {"id": thing_id}
 
 
class DomainError(Exception):
    def __init__(self, code: str, status: int = 400):
        self.code, self.status = code, status
 
 
@app.exception_handler(DomainError)
async def domain_error(
    request: Request, exc: DomainError,
) -> JSONResponse:
    return JSONResponse(
        {"error": exc.code}, status_code=exc.status,
    )
 
 
@app.exception_handler(RequestValidationError)
async def invalid(
    request: Request, exc: RequestValidationError,
) -> JSONResponse:
    fields = [".".join(map(str, e["loc"]))
              for e in exc.errors()]
    return JSONResponse(
        {"error": "invalid", "fields": fields},
        status_code=422,
    )

from collections.abc import AsyncIterator
from typing import Annotated
 
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import APIKeyHeader
 
 
async def get_db() -> AsyncIterator[Session]:  # yield dep
    db = Session()
    try:
        yield db                # value injected
        await db.commit()       # after the endpoint returns
    except Exception:
        await db.rollback()
        raise                   # always re-raise
    finally:
        await db.close()
 
 
class Paging:                   # class dep: __init__ params
    def __init__(self, limit: int = 20, offset: int = 0):
        self.limit = min(limit, 100)
        self.offset = offset
 
 
api_key = APIKeyHeader(name="X-API-Key", auto_error=False)
 
 
async def require_key(
    key: Annotated[str | None, Depends(api_key)],  # sub-dep
) -> str:
    if key != "secret":
        raise HTTPException(401, "Bad API key")
    return key
 
 
DbDep = Annotated[Session, Depends(get_db)]
router = APIRouter(dependencies=[Depends(require_key)])
 
 
@router.get("/items")
async def list_items(
    db: DbDep, page: Annotated[Paging, Depends()],
) -> list[Item]:
    return await db.list_items(page.limit, page.offset)

from fastapi import APIRouter
 
router = APIRouter(
    prefix="/items",
    tags=["items"],
    responses={404: {"description": "Not found"}},
)
 
 
@router.get("/")
async def list_items() -> list[str]:
    return []

from fastapi import APIRouter, FastAPI
 
from app.routers import items, users
 
api = APIRouter(prefix="/api/v1")
api.include_router(users.router)
api.include_router(items.router)
 
app = FastAPI(title="Notes API")
app.include_router(api)

from fastapi import BackgroundTasks
 
 
def send_welcome(email: str) -> None:  # sync: threadpool
    ...
 
 
@app.post("/signup", status_code=202)
async def signup(
    email: EmailStr, tasks: BackgroundTasks,
) -> dict[str, str]:
    tasks.add_task(send_welcome, email)  # after response
    return {"status": "queued"}

from collections.abc import (
    AsyncIterator, Awaitable, Callable,
)
from contextlib import asynccontextmanager
 
import httpx
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
 
 
@asynccontextmanager
async def lifespan(
    app: FastAPI,
) -> AsyncIterator[dict[str, object]]:
    client = httpx.AsyncClient(timeout=10)  # startup
    yield {"http": client}          # -> request.state.http
    await client.aclose()           # shutdown
 
 
app = FastAPI(lifespan=lifespan)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Request-ID"],
)
 
 
@app.middleware("http")
async def no_cache(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    response = await call_next(request)
    response.headers["Cache-Control"] = "no-store"
    return response

uv add pyjwt "pwdlib[argon2]"

from datetime import UTC, datetime, timedelta
from typing import Annotated
 
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer, OAuth2PasswordRequestForm,
)
from pwdlib import PasswordHash
 
hasher = PasswordHash.recommended()        # argon2id
oauth2 = OAuth2PasswordBearer(tokenUrl="/token")
 
 
def make_token(sub: str, secret: str) -> str:
    exp = datetime.now(UTC) + timedelta(minutes=30)
    return jwt.encode(
        {"sub": sub, "exp": exp}, secret, algorithm="HS256",
    )
 
 
@app.post("/token")
async def login(
    form: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DbDep, settings: SettingsDep,
) -> Token:
    user = await db.user_by_email(form.username)
    if not user or not hasher.verify(
        form.password, user.password_hash,
    ):
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            "Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    secret = settings.jwt_secret.get_secret_value()
    return Token(access_token=make_token(user.email, secret))

from functools import lru_cache
from typing import Annotated
 
from fastapi import Depends
from pydantic import PostgresDsn, SecretStr
from pydantic_settings import (
    BaseSettings, SettingsConfigDict,
)
 
 
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="APP_",         # APP_DEBUG=true
        env_nested_delimiter="__",
        extra="ignore",
    )
    debug: bool = False
    database_url: PostgresDsn      # required
    jwt_secret: SecretStr          # prints as **********
    cors_origins: list[str] = []   # JSON: '["https://a.io"]'
 
 
@lru_cache
def get_settings() -> Settings:
    return Settings()  # type: ignore[call-arg]  # reads env
 
 
SettingsDep = Annotated[Settings, Depends(get_settings)]

uv add sqlmodel "sqlalchemy[asyncio]" asyncpg  # or psycopg
uv add alembic                                 # migrations

from collections.abc import AsyncIterator
from typing import Annotated
 
from fastapi import Depends
from sqlalchemy.ext.asyncio import (
    async_sessionmaker, create_async_engine,
)
from sqlmodel import Field, SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
 
engine = create_async_engine(
    "postgresql+asyncpg://app:secret@localhost/app",
    pool_size=10,
    pool_pre_ping=True,
)
Session = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False,
)
 
 
async def get_session() -> AsyncIterator[AsyncSession]:
    async with Session() as session:
        yield session
 
 
SessionDep = Annotated[AsyncSession, Depends(get_session)]
 
 
class HeroBase(SQLModel):
    name: str = Field(index=True, max_length=100)
    age: int | None = Field(default=None, ge=0)
 
 
class Hero(HeroBase, table=True):      # the table
    id: int | None = Field(default=None, primary_key=True)
 
 
class HeroCreate(HeroBase): ...         # request body
 
 
class HeroUpdate(SQLModel):            # PATCH body
    name: str | None = None
    age: int | None = None
 
 
class HeroPublic(HeroBase):            # response
    id: int

app = FastAPI(
    title="Notes API",
    version="1.2.0",
    summary="Personal notes service",
    openapi_tags=[{"name": "notes", "description": "CRUD"}],
    docs_url="/docs",             # None disables Swagger UI
    redoc_url=None,
    openapi_url="/openapi.json",  # None disables all docs
    swagger_ui_parameters={"persistAuthorization": True},
)
 
 
@app.get(
    "/notes/{note_id}",
    tags=["notes"],
    summary="Get one note",
    response_description="The note",
    responses={404: {"model": ErrorOut}},
    operation_id="getNote",
    deprecated=False,
)
async def get_note(note_id: int) -> NoteOut:
    """Markdown docstring shows up as the description."""
    ...

uv add --dev pytest httpx
uv run pytest -q

from fastapi.testclient import TestClient
 
from app.main import app
 
 
def test_health() -> None:
    with TestClient(app) as client:     # runs lifespan
        r = client.get("/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}

import pytest
from httpx import ASGITransport, AsyncClient
 
from app.main import app
 
 
@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"
 
 
@pytest.mark.anyio
async def test_health_async() -> None:
    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport, base_url="http://test",
    ) as ac:
        r = await ac.get("/health")
    assert r.status_code == 200

FROM python:3.14-slim
COPY --from=ghcr.io/astral-sh/uv:0.12.19 /uv /bin/
WORKDIR /app
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy
COPY pyproject.toml uv.lock ./
RUN uv sync --locked --no-dev --no-install-project
COPY app ./app
RUN uv sync --locked --no-dev
USER nobody
EXPOSE 8000
CMD ["/app/.venv/bin/fastapi", "run", "app/main.py", \
     "--port", "8000"]

router = APIRouter(prefix="/heroes", tags=["heroes"])
 
 
@router.post("/", status_code=201)
async def create(
    body: HeroCreate, s: SessionDep,
) -> HeroPublic:
    hero = Hero.model_validate(body)
    s.add(hero)
    await s.commit()
    await s.refresh(hero)
    return HeroPublic.model_validate(hero)
 
 
async def get_or_404(hero_id: int, s: SessionDep) -> Hero:
    if (hero := await s.get(Hero, hero_id)) is None:
        raise HTTPException(404, "Hero not found")
    return hero
 
 
HeroDep = Annotated[Hero, Depends(get_or_404)]
 
 
@router.get("/{hero_id}")
async def read(hero: HeroDep) -> HeroPublic:
    return HeroPublic.model_validate(hero)
 
 
@router.patch("/{hero_id}")
async def update(
    hero: HeroDep, body: HeroUpdate, s: SessionDep,
) -> HeroPublic:
    hero.sqlmodel_update(body.model_dump(exclude_unset=True))
    await s.commit()
    await s.refresh(hero)
    return HeroPublic.model_validate(hero)
 
 
@router.delete("/{hero_id}", status_code=204)
async def delete(hero: HeroDep, s: SessionDep) -> None:
    await s.delete(hero)
    await s.commit()

from typing import Annotated
 
from fastapi import Query
from pydantic import BaseModel, Field
from sqlmodel import col, func, select
 
 
class PageParams(BaseModel):
    limit: int = Field(20, ge=1, le=100)
    offset: int = Field(0, ge=0)
 
 
class Page[T](BaseModel):
    items: list[T]
    total: int
 
 
PageDep = Annotated[PageParams, Query()]
 
 
@router.get("/")
async def list_heroes(
    p: PageDep, s: SessionDep,
) -> Page[HeroPublic]:
    total = (await s.exec(
        select(func.count()).select_from(Hero))).one()
    rows = await s.exec(select(Hero).order_by(col(Hero.id))
                        .offset(p.offset).limit(p.limit))
    items = [HeroPublic.model_validate(h) for h in rows]
    return Page(items=items, total=total)

import time
import uuid
 
 
@app.middleware("http")
async def request_context(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    rid = request.headers.get("x-request-id")
    rid = rid or str(uuid.uuid4())
    request.state.request_id = rid
    start = time.perf_counter()
    response = await call_next(request)
    ms = (time.perf_counter() - start) * 1000
    response.headers["X-Request-ID"] = rid
    response.headers["Server-Timing"] = f"app;dur={ms:.1f}"
    return response

async def current_user(
    token: Annotated[str, Depends(oauth2)],
    db: DbDep, settings: SettingsDep,
) -> User:
    denied = HTTPException(401, "Invalid token",
                           {"WWW-Authenticate": "Bearer"})
    key = settings.jwt_secret.get_secret_value()
    try:
        claims = jwt.decode(token, key, algorithms=["HS256"])
    except jwt.InvalidTokenError:
        raise denied from None
    user = await db.user_by_email(claims.get("sub", ""))
    if user is None or user.disabled:
        raise denied
    return user
 
 
CurrentUser = Annotated[User, Depends(current_user)]
 
 
def require_role(role: str) -> Callable[[User], User]:
    def check(user: CurrentUser) -> User:
        if role not in user.roles:
            raise HTTPException(403, "Forbidden")
        return user
    return check
 
 
@app.delete("/admin/cache", dependencies=[
    Depends(require_role("admin"))])
async def clear_cache() -> None: ...

from fastapi import File, Form, UploadFile
 
MAX_BYTES = 5 * 1024 * 1024
 
 
@app.post("/avatars", status_code=201)
async def upload_avatar(
    file: Annotated[UploadFile, File()],
    alt: Annotated[str, Form(max_length=200)] = "",
) -> dict[str, object]:
    if file.content_type not in {"image/png", "image/jpeg"}:
        raise HTTPException(415, "PNG or JPEG only")
    data = await file.read(MAX_BYTES + 1)
    if len(data) > MAX_BYTES:
        raise HTTPException(413, "Max 5 MB")
    name = file.filename or "upload"
    return {"name": name, "size": len(data), "alt": alt}

from collections.abc import AsyncIterable
 
import anyio
from fastapi.sse import EventSourceResponse, ServerSentEvent
 
 
class Progress(BaseModel):
    step: int
    done: bool
 
 
@app.get("/jobs/{job_id}/events",
         response_class=EventSourceResponse)
async def job_events(
    job_id: str,
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    start = (last_event_id or 0) + 1
    for step in range(start, 6):
        await anyio.sleep(1)
        yield ServerSentEvent(
            data=Progress(step=step, done=step == 5),
            event="progress", id=str(step),
        )

from collections.abc import Iterator
 
import pytest
from fastapi.testclient import TestClient
 
from app.deps import current_user, get_session
from app.main import app
from app.models import User
 
 
@pytest.fixture
def client(session: FakeSession) -> Iterator[TestClient]:
    app.dependency_overrides[current_user] = lambda: User(
        id=1, email="t@example.com", roles=["admin"],
    )
    app.dependency_overrides[get_session] = lambda: session
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()