FastAPI Cheatsheet 03 — Response Models, Serialization, and Status Codes

class UserOut(BaseModel):
    id: int
    email: EmailStr
    full_name: str

@app.get("/users/{id}", response_model=UserOut)
async def get_(id: int):
    user = await db.get_user(id)
    return user            # SQLAlchemy / dict / Pydantic — Pydantic shapes it

@app.get(
    "/users/{id}",
    response_model=UserOut,
    response_model_exclude={"email"},
    response_model_exclude_none=True,
    response_model_exclude_unset=True,
    response_model_exclude_defaults=False,
)

class UserOut(BaseModel):
    id: int
    email: str
    model_config = {"from_attributes": True}

from fastapi import status

@app.post("/users", status_code=status.HTTP_201_CREATED, response_model=UserOut)
@app.delete("/users/{id}", status_code=204)        # no body

from fastapi import Response

@app.post("/login")
async def login(response: Response):
    response.set_cookie("sid", "abc", httponly=True, secure=True, samesite="lax", max_age=86400)
    response.headers["x-custom"] = "v"
    return {"ok": True}

from fastapi.responses import (
    JSONResponse, ORJSONResponse, UJSONResponse,
    HTMLResponse, PlainTextResponse, RedirectResponse,
    StreamingResponse, FileResponse,
)

# Default
app = FastAPI(default_response_class=ORJSONResponse)   # faster JSON

@app.get("/", response_class=HTMLResponse)
async def home():
    return "<h1>Hi</h1>"

@app.get("/legacy")
async def redir():
    return RedirectResponse(url="/", status_code=308)

@app.get("/stream")
async def stream():
    async def gen():
        for i in range(10):
            yield f"chunk {i}\n".encode()
    return StreamingResponse(gen(), media_type="text/plain")

@app.get("/events")
async def events(request: Request):
    async def gen():
        while True:
            if await request.is_disconnected(): break
            yield f"data: {json.dumps({'t': time.time()})}\n\n"
            await asyncio.sleep(15)
    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

@app.get("/dl/{name}")
async def dl(name: str):
    return FileResponse(f"/files/{name}", filename=name, media_type="application/octet-stream")

@app.get(
    "/users/{id}",
    response_model=UserOut,
    responses={
        404: {"description": "Not found", "model": ErrorOut},
        429: {"description": "Rate limited"},
    },
)
async def get_(id: int): ...

from typing import TypeVar, Generic
T = TypeVar("T")

class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    has_more: bool

@app.get("/users", response_model=Page[UserOut])
async def list_(): ...

from pydantic import field_serializer

class Event(BaseModel):
    when: datetime
    @field_serializer("when", when_used="json")
    def to_iso(self, v, _info): return v.isoformat()

@app.get("/posts")
async def posts(response: Response):
    response.headers["Cache-Control"] = "public, max-age=60, s-maxage=120"
    return await db.list_posts()

@app.get("/u/{id}")
async def u(id: int, if_none_match: str | None = Header(None)):
    user = await db.get_user(id)
    etag = compute_etag(user)
    if if_none_match == etag:
        return Response(status_code=304)
    return JSONResponse(user.model_dump(), headers={"ETag": etag})