Python / FastAPI Basics Interview Questions

# main.py
from fastapi import FastAPI

app = FastAPI()

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

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

# Run the server
uvicorn main:app --reload

# main  = the module (main.py)
# app   = the FastAPI() instance variable
# --reload = auto-restart on code changes (dev only)

from fastapi import FastAPI
from enum import Enum

app = FastAPI()

# PATH parameter: part of the URL — /users/42
@app.get("/users/{user_id}")
def get_user(user_id: int):  # int type enforced automatically
    return {"user_id": user_id}

# QUERY parameters: after ? in URL — /items?skip=0&limit=10
@app.get("/items")
def list_items(
    skip: int = 0,          # optional, default 0
    limit: int = 10,        # optional, default 10
    search: str | None = None,  # fully optional
):
    return {"skip": skip, "limit": limit, "search": search}

# Enum path param — validates allowed values
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet  = "resnet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    return {"model": model_name}

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Annotated
from pydantic import Field

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None  # optional field
    price: float
    tax: float | None = None

@app.post("/items")
def create_item(item: Item):  # body parsed + validated automatically
    total = item.price + (item.tax or 0)
    return {**item.model_dump(), "total_price": total}

# Mixing path param + query param + body in the same endpoint
@app.put("/items/{item_id}")
def update_item(
    item_id: int,          # path
    q: str | None = None,  # query
    item: Item | None = None,  # body (optional)
):
    result = {"item_id": item_id}
    if q:
        result["q"] = q
    if item:
        result.update(item.model_dump())
    return result

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Annotated
from datetime import datetime

class UserCreate(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email:    Annotated[str, Field(pattern=r"^[\w.-]+@[\w.-]+\.\w+$")]
    age:      Annotated[int, Field(ge=18, le=120)]  # >= 18 and <= 120
    password: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

    # Field-level validator
    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("Username must be alphanumeric")
        return v.lower()

    # Cross-field validator (model level)
    @model_validator(mode="after")
    def passwords_match(self) -> "UserCreate":
        # example: confirm_password field check would go here
        return self

class UserRead(BaseModel):  # separate model for responses (no password)
    username: str
    email: str
    age: int
    created_at: datetime

    model_config = {"from_attributes": True}  # allows ORM object input

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: str

class UserOut(BaseModel):
    username: str
    email: str
    # password intentionally excluded

# response_model filters the output — password never appears in response
@app.post("/users", response_model=UserOut)
def create_user(user: UserIn) -> UserOut:
    # Even if we return the full user object, password is stripped
    return user  # FastAPI applies UserOut filtering

# response_model_exclude_unset: only include fields explicitly set by caller
@app.get("/items/{id}", response_model=UserOut, response_model_exclude_unset=True)
def get_item(id: int):
    return {"username": "alice", "email": "a@b.com"}

# List response
@app.get("/users", response_model=list[UserOut])
def list_users():
    return [{"username": "alice", "email": "a@b.com", "password": "secret"}]

from fastapi import FastAPI, Path, Query
from typing import Annotated

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(
        title="The ID of the item",
        description="Must be a positive integer",
        ge=1,           # >= 1
        le=1_000_000,   # <= 1,000,000
    )],
    q: Annotated[str | None, Query(
        min_length=3,
        max_length=50,
        pattern=r"^[a-z]+$",
        alias="search",     # URL uses ?search=... instead of ?q=...
        deprecated=True,    # marks param as deprecated in docs
    )] = None,
):
    return {"item_id": item_id, "q": q}

# Required query param with no default
@app.get("/search")
def search(
    q: Annotated[str, Query(min_length=1)],  # required — no default
):
    return {"query": q}

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse, Response, FileResponse

app = FastAPI()

fake_db = {1: {"name": "Foo"}, 2: {"name": "Bar"}}

# Set default success status code
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    return {"name": name}

# Raise HTTP errors
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found",
            headers={"X-Error": "Item missing"},  # custom header on error
        )
    return fake_db[item_id]

# Custom JSONResponse for full control
@app.get("/custom")
def custom_response():
    return JSONResponse(
        status_code=200,
        content={"message": "ok"},
        headers={"X-Custom": "value"},
    )

# 204 No Content (no response body)
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
    fake_db.pop(item_id, None)
    return Response(status_code=204)

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

app = FastAPI()

# Simple dependency — shared pagination logic
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

@app.get("/items")
def list_items(pagination: Annotated[dict, Depends(pagination)]):
    return {"pagination": pagination}

@app.get("/users")
def list_users(pagination: Annotated[dict, Depends(pagination)]):
    return {"pagination": pagination}

# Class-based dependency
class DBSession:
    def __init__(self):
        self.db = "fake_db_connection"  # in real code: create session

    def close(self):
        pass  # close connection

# Generator dependency — enables cleanup with finally
def get_db():
    db = DBSession()
    try:
        yield db.db   # yield makes it a context-managed dependency
    finally:
        db.close()    # runs after request completes, even on error

@app.get("/data")
def get_data(db: Annotated[str, Depends(get_db)]):
    return {"db": db}

# Nested dependencies
def verify_token(token: str) -> str:
    if token != "secret":
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

def get_current_user(token: Annotated[str, Depends(verify_token)]):
    return {"user": "alice", "token": token}

# routers/items.py
from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/items",          # all routes here are prefixed
    tags=["items"],           # Swagger UI grouping label
    responses={404: {"description": "Not found"}},  # shared response docs
)

@router.get("/")
def list_items():
    return [{"name": "item1"}]

@router.get("/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id}

@router.post("/", status_code=201)
def create_item(name: str):
    return {"name": name}

# routers/users.py
from fastapi import APIRouter

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

@router.get("/")
def list_users():
    return [{"username": "alice"}]

# main.py
from fastapi import FastAPI
from routers import items, users

app = FastAPI()

app.include_router(items.router)
app.include_router(users.router)
app.include_router(
    users.router,
    prefix="/v2",         # override prefix for a second version
    dependencies=[Depends(verify_admin)],  # apply dep to all routes
)

@app.get("/")  # root route stays in main.py
def root():
    return {"message": "API root"}

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

app = FastAPI()

# Custom middleware using decorator
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)  # passes request to next handler
    duration = time.time() - start
    response.headers["X-Process-Time"] = str(duration)
    return response

# CORS — allow cross-origin requests (e.g. from a React frontend)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myfrontend.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip compression for large responses
app.add_middleware(GZipMiddleware, minimum_size=1000)

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# GOOD: truly async I/O — does not block the event loop
@app.get("/async-data")
async def get_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
    return response.json()

# ALSO GOOD: synchronous — FastAPI runs this in a thread pool
@app.get("/sync-data")
def get_data_sync():
    import requests  # sync library
    response = requests.get("https://api.example.com/data")
    return response.json()

# DANGEROUS: blocking call inside async def without await
# This blocks the entire event loop, killing concurrency
@app.get("/bad")
async def bad_handler():
    import time
    time.sleep(5)  # NEVER do this in async def
    return {"message": "slow"}

# CORRECT: use asyncio.sleep in async context
@app.get("/good-async-wait")
async def good_handler():
    await asyncio.sleep(5)  # non-blocking
    return {"message": "waited"}

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

def send_welcome_email(email: str, username: str):
    # Simulate sending an email (could call an email service)
    print(f"Sending welcome email to {email} for {username}")

def write_audit_log(action: str, user_id: int):
    print(f"Audit: user {user_id} performed {action}")

class UserIn(BaseModel):
    username: str
    email: str

@app.post("/register", status_code=201)
def register_user(user: UserIn, background_tasks: BackgroundTasks):
    # Response is returned immediately
    # Email is sent after the response
    background_tasks.add_task(
        send_welcome_email,
        email=user.email,
        username=user.username,
    )
    background_tasks.add_task(write_audit_log, "register", user_id=42)
    return {"message": f"User {user.username} created"}

# BackgroundTasks can also be injected via dependencies
def get_background(background_tasks: BackgroundTasks) -> BackgroundTasks:
    return background_tasks

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from typing import Annotated
from datetime import datetime, timedelta
import jwt  # pip install PyJWT

app = FastAPI()
SECRET_KEY = "your-secret-key"  # use a strong random key in production!
ALGORITHM = "HS256"

# Tells FastAPI where clients obtain tokens — used in OpenAPI docs
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

# 1. Login endpoint — returns a JWT
@app.post("/token")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    if form.username != "alice" or form.password != "secret":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # Create JWT token
    payload = {
        "sub": form.username,
        "exp": datetime.utcnow() + timedelta(minutes=30),
    }
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": token, "token_type": "bearer"}

# 2. Dependency that decodes and validates the token
def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise ValueError
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return username

# 3. Protected endpoint
@app.get("/me")
def read_me(current_user: Annotated[str, Depends(get_current_user)]):
    return {"username": current_user}

from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated
from enum import Enum

app = FastAPI()

class Role(str, Enum):
    user  = "user"
    admin = "admin"

class User:
    def __init__(self, username: str, role: Role):
        self.username = username
        self.role = role

# Simulated auth — in reality decode a JWT
def get_current_user() -> User:
    return User(username="alice", role=Role.user)

# Higher-order dependency factory — creates a role checker
def require_role(role: Role):
    def checker(user: Annotated[User, Depends(get_current_user)]) -> User:
        if user.role != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Requires {role} role",
            )
        return user
    return checker

# Public route — any authenticated user
@app.get("/profile")
def profile(user: Annotated[User, Depends(get_current_user)]):
    return {"username": user.username, "role": user.role}

# Admin-only route
@app.delete("/users/{user_id}")
def delete_user(
    user_id: int,
    _: Annotated[User, Depends(require_role(Role.admin))],
):
    return {"deleted": user_id}

# Router-level dependency — apply to all routes in a router
from fastapi import APIRouter
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(require_role(Role.admin))],
)

@admin_router.get("/stats")
def admin_stats():
    return {"total_users": 42}

# pip install sqlalchemy asyncpg
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select
from fastapi import FastAPI, Depends
from typing import Annotated

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id:       Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email:    Mapped[str]

# DB dependency — yields session, always closes it
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

DBDep = Annotated[AsyncSession, Depends(get_db)]

app = FastAPI()

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

@app.get("/users")
async def list_users(db: DBDep):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return users

@app.post("/users", status_code=201)
async def create_user(username: str, email: str, db: DBDep):
    user = User(username=username, email=email)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

# 1. Install and initialise Alembic
# pip install alembic
# alembic init alembic
# This creates alembic/ directory and alembic.ini

# alembic/env.py — point to your models and database
from app.database import Base, DATABASE_URL  # your app imports
from app import models  # noqa: import models so Alembic can see them

# In run_migrations_online():
connectable = create_engine(DATABASE_URL)
# Set target_metadata so Alembic compares your models vs the DB
target_metadata = Base.metadata

# 2. Generate a migration after changing a model
alembic revision --autogenerate -m "add users table"
# Creates alembic/versions/xxxx_add_users_table.py

# 3. Apply migrations
alembic upgrade head     # apply all pending migrations
alembic upgrade +1       # apply one migration

# 4. Rollback
alembic downgrade -1     # undo one migration
alembic downgrade base   # undo ALL migrations

# 5. View history
alembic history          # list all migrations
alembic current          # show current DB revision

# Generated migration file — always review before applying!
def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("username", sa.String(), nullable=False),
        sa.Column("email", sa.String(), nullable=False),
    )

def downgrade() -> None:
    op.drop_table("users")

# app/main.py
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

def get_db():
    return {"connection": "real_db"}

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
def root():
    return {"message": "Hello"}

@app.post("/items", status_code=201)
def create_item(item: Item, db=Depends(get_db)):
    return {"created": item.name, "db": db["connection"]}

# tests/test_main.py
import pytest
from fastapi.testclient import TestClient
from app.main import app, get_db

# Override the real DB dependency with a mock
def override_get_db():
    return {"connection": "test_db"}

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello"}

def test_create_item():
    response = client.post(
        "/items",
        json={"name": "Widget", "price": 9.99},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["created"] == "Widget"
    assert data["db"] == "test_db"  # mock dependency used

def test_create_item_invalid():
    response = client.post("/items", json={"name": "Widget"})  # missing price
    assert response.status_code == 422  # validation error

# Async test with httpx.AsyncClient
import pytest
from httpx import AsyncClient, ASGITransport

@pytest.mark.asyncio
async def test_root_async():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# Custom exception class
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Handler for our custom exception
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": "not_found", "item_id": exc.item_id},
    )

# Override the default validation error handler
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_failed",
            "detail": exc.errors(),
            "body": exc.body,
        },
    )

# Override generic HTTP exception handler
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": "http_error", "detail": exc.detail},
    )

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id == 999:
        raise ItemNotFoundError(item_id=item_id)  # triggers custom handler
    return {"item_id": item_id}

from fastapi import FastAPI, Form, File, UploadFile
from typing import Annotated

app = FastAPI()

# Form data (application/x-www-form-urlencoded)
@app.post("/login")
def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
):
    return {"username": username}

# Single file upload
@app.post("/upload")
async def upload_file(file: UploadFile):
    contents = await file.read()  # bytes
    return {
        "filename":     file.filename,
        "content_type": file.content_type,
        "size":         len(contents),
    }

# Multiple files
@app.post("/upload-multiple")
async def upload_files(files: list[UploadFile]):
    return [{"filename": f.filename} for f in files]

# Mix form fields + file
@app.post("/profile")
async def update_profile(
    username: Annotated[str, Form()],
    avatar:   UploadFile | None = None,
):
    result = {"username": username}
    if avatar:
        content = await avatar.read()
        # Save to disk / cloud storage
        result["avatar_size"] = len(content)
    return result

# Validate file type and size
@app.post("/images")
async def upload_image(file: UploadFile):
    if file.content_type not in ["image/jpeg", "image/png"]:
        from fastapi import HTTPException
        raise HTTPException(400, "Only JPEG/PNG allowed")
    content = await file.read()
    if len(content) > 5 * 1024 * 1024:  # 5 MB limit
        raise HTTPException(400, "File too large")
    return {"filename": file.filename}

# pip install pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache
from fastapi import FastAPI, Depends
from typing import Annotated

class Settings(BaseSettings):
    app_name:    str = "My API"
    debug:       bool = False
    db_url:      str = "sqlite:///./test.db"
    secret_key:  str
    jwt_expire_minutes: int = 30

    model_config = SettingsConfigDict(
        env_file=".env",          # read from .env file
        env_file_encoding="utf-8",
        case_sensitive=False,     # DB_URL and db_url both work
    )

@lru_cache  # singleton — reads .env once, cached for app lifetime
def get_settings() -> Settings:
    return Settings()

SettingsDep = Annotated[Settings, Depends(get_settings)]

app = FastAPI()

@app.get("/info")
def app_info(settings: SettingsDep):
    return {"app_name": settings.app_name, "debug": settings.debug}

# .env file (never commit to git)
SECRET_KEY=my-super-secret-key-here
DB_URL=postgresql+asyncpg://user:pass@localhost/mydb
DEBUG=false

# tests — override settings easily
from app.main import app, get_settings
from app.config import Settings

def override_settings():
    return Settings(secret_key="test-key", db_url="sqlite:///./test.db")

app.dependency_overrides[get_settings] = override_settings

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg

# Module-level storage for shared resources
db_pool: asyncpg.Pool | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- STARTUP: runs before the app starts serving requests ---
    global db_pool
    db_pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db",
        min_size=5,
        max_size=20,
    )
    print("Database pool created")

    yield  # app runs here

    # --- SHUTDOWN: runs after last request is processed ---
    await db_pool.close()
    print("Database pool closed")

app = FastAPI(lifespan=lifespan)

@app.get("/users")
async def list_users():
    async with db_pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, username FROM users")
    return [dict(r) for r in rows]

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

# Simple echo WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

# Connection manager for broadcast (e.g. chat room)
class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/chat/{room}")
async def chat(websocket: WebSocket, room: str):
    await manager.connect(websocket)
    try:
        while True:
            msg = await websocket.receive_text()
            await manager.broadcast(f"[{room}] {msg}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"A user left {room}")

# Dockerfile
FROM python:3.12-slim AS base
WORKDIR /app

# Install dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user for security
RUN adduser --disabled-password --gecos "" appuser
USER appuser

EXPOSE 8000

# Production: Gunicorn manages multiple Uvicorn worker processes
CMD ["gunicorn", "main:app",\
     "--workers", "4",\
     "--worker-class", "uvicorn.workers.UvicornWorker",\
     "--bind", "0.0.0.0:8000"]

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db/mydb
      - SECRET_KEY=${SECRET_KEY}  # from .env
    depends_on:
      db:
        condition: service_healthy
    restart: unless-stopped

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 5s
      retries: 5
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

# Health check endpoint
from fastapi import FastAPI
from sqlalchemy import text

app = FastAPI()

@app.get("/health")
async def health(db: DBDep):
    try:
        await db.execute(text("SELECT 1"))
        db_ok = True
    except Exception:
        db_ok = False
    return {
        "status": "healthy" if db_ok else "degraded",
        "database": "ok" if db_ok else "error",
    }

# Nginx config snippet for reverse proxy
# server {
#     listen 443 ssl;
#     server_name api.example.com;
#     location / {
#         proxy_pass http://localhost:8000;
#         proxy_set_header Host $host;
#         proxy_set_header X-Real-IP $remote_addr;
#     }
# }

# slowapi — rate limiting middleware
# pip install slowapi
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/limited")
@limiter.limit("10/minute")
async def limited_route(request: Request):
    return {"message": "ok"}

from pydantic import BaseModel, field_validator, model_validator, ConfigDict
from typing import Self

class OrderItem(BaseModel):
    product_id: int
    quantity:   int
    unit_price: float

    @field_validator("quantity")
    @classmethod
    def quantity_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("quantity must be positive")
        return v

class Order(BaseModel):
    items: list[OrderItem]
    discount: float = 0.0

    model_config = ConfigDict(from_attributes=True)  # replaces orm_mode

    @model_validator(mode="after")
    def discount_valid(self) -> Self:
        if self.discount < 0 or self.discount > 1:
            raise ValueError("discount must be between 0 and 1")
        return self

    @property
    def total(self) -> float:
        subtotal = sum(i.quantity * i.unit_price for i in self.items)
        return round(subtotal * (1 - self.discount), 2)

# Serialisation
order = Order(items=[OrderItem(product_id=1, quantity=2, unit_price=9.99)])
print(order.model_dump())        # dict
print(order.model_dump_json())   # JSON bytes
print(order.model_json_schema()) # JSON Schema

# Option 1: In-process LRU cache for slow dependencies
from functools import lru_cache
from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

@lru_cache(maxsize=128)
def get_config_from_db(key: str) -> str:
    # Called once per unique key, then cached in memory
    return f"value_for_{key}"

@app.get("/config/{key}")
def get_config(key: str):
    return {"value": get_config_from_db(key)}

# Option 2: Redis cache with fastapi-cache2
# pip install fastapi-cache2[redis]
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    redis = aioredis.from_url("redis://localhost")
    FastAPICache.init(RedisBackend(redis), prefix="myapp-cache")
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/expensive")
@cache(expire=60)  # cache for 60 seconds in Redis
async def expensive_operation():
    import asyncio
    await asyncio.sleep(2)  # simulate slow DB query
    return {"result": "computed"}

# Option 3: HTTP Cache-Control headers
from fastapi.responses import Response

@app.get("/public-data")
def public_data(response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600"
    return {"data": "cacheable by CDN and browsers"}

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from pydantic import BaseModel, Field

# App-level metadata
app = FastAPI(
    title="My Inventory API",
    description="## Manage your warehouse inventory\n\nSupports CRUD for items and categories.",
    version="2.1.0",
    contact={"name": "Dev Team", "email": "dev@example.com"},
    license_info={"name": "MIT"},
    terms_of_service="https://example.com/terms",
    docs_url="/docs",    # Swagger UI path
    redoc_url="/redoc",  # ReDoc path
    openapi_url="/openapi.json",
)

class Item(BaseModel):
    name: str = Field(
        ...,
        examples=["Widget"],
        description="Human-readable item name",
    )
    price: float = Field(
        ..., gt=0, examples=[9.99],
        description="Price in USD, must be positive",
    )

@app.post(
    "/items",
    summary="Create a new inventory item",
    description="Creates an item and returns it with its assigned ID.",
    response_description="The created item including its database ID",
    tags=["inventory"],
    status_code=201,
)
def create_item(item: Item) -> Item:
    return item

# Disable docs in production (optional)
# app = FastAPI(docs_url=None, redoc_url=None)

# pip install celery redis
# celery_app.py
from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
celery_app.conf.task_serializer = "json"

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, email: str, subject: str, body: str):
    try:
        # call email service
        print(f"Sending {subject} to {email}")
    except Exception as exc:
        raise self.retry(exc=exc)  # retry up to 3 times

# main.py — FastAPI triggers Celery tasks
from fastapi import FastAPI
from pydantic import BaseModel
from celery_app import send_email

app = FastAPI()

class EmailRequest(BaseModel):
    to:      str
    subject: str
    body:    str

@app.post("/send-email", status_code=202)
def trigger_email(req: EmailRequest):
    # .delay() sends the task to the broker queue
    task = send_email.delay(req.to, req.subject, req.body)
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
def task_status(task_id: str):
    from celery.result import AsyncResult
    result = AsyncResult(task_id)
    return {"task_id": task_id, "status": result.status}

# Run the Celery worker (separate process)
# celery -A celery_app worker --loglevel=info

# 1. Measure with locust load testing
# pip install locust
# locustfile.py
from locust import HttpUser, task, between

class APIUser(HttpUser):
    wait_time = between(0.1, 1)

    @task(3)
    def list_items(self):
        self.client.get("/items")

    @task(1)
    def create_item(self):
        self.client.post("/items", json={"name": "test", "price": 1.0})

# locust -f locustfile.py --host=http://localhost:8000

# 2. Add request timing middleware
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Response-Time"] = f"{elapsed:.4f}s"
    if elapsed > 0.5:  # log slow requests
        print(f"SLOW: {request.url.path} took {elapsed:.3f}s")
    return response

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

app = FastAPI()

# Class-based dependency — holds configuration
class PaginationParams:
    def __init__(
        self,
        skip:  int = 0,
        limit: int = 10,
        max_limit: int = 100,
    ):
        if limit > max_limit:
            raise HTTPException(400, f"limit cannot exceed {max_limit}")
        self.skip  = skip
        self.limit = limit

# Sub-dependency tree:
# get_current_user → verify_token → oauth2_scheme
def verify_token(token: str = Depends(oauth2_scheme)) -> dict:
    if token != "valid":
        raise HTTPException(401, "Bad token")
    return {"username": "alice", "role": "admin"}

def get_current_user(
    payload: dict = Depends(verify_token),
) -> dict:
    return {"username": payload["username"]}

Pagination = Annotated[PaginationParams, Depends(PaginationParams)]
CurrentUser = Annotated[dict, Depends(get_current_user)]

@app.get("/items")
def list_items(page: Pagination, user: CurrentUser):
    return {
        "user":  user["username"],
        "skip":  page.skip,
        "limit": page.limit,
    }

# Dependency with use_cache=False — new instance per call
@app.get("/no-cache")
def no_cache(
    a: Annotated[PaginationParams, Depends(PaginationParams, use_cache=False)],
    b: Annotated[PaginationParams, Depends(PaginationParams, use_cache=False)],
):
    # a and b are separate instances (default: same instance per request)
    return {"a_skip": a.skip, "b_skip": b.skip}

# pip install pytest pytest-asyncio httpx
# conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app, get_db
from app.database import AsyncSessionLocal, Base, engine

@pytest.fixture(scope="function")
async def async_client():
    # Use a test database
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async def override_db():
        async with AsyncSessionLocal() as session:
            yield session

    app.dependency_overrides[get_db] = override_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client

    app.dependency_overrides.clear()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

# tests/test_users.py
import pytest

@pytest.mark.asyncio
async def test_create_user(async_client):
    response = await async_client.post(
        "/users",
        json={"username": "alice", "email": "alice@example.com"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "alice"
    assert "password" not in data  # response_model excludes it

@pytest.mark.asyncio
async def test_list_users_empty(async_client):
    response = await async_client.get("/users")
    assert response.status_code == 200
    assert response.json() == []

@pytest.mark.asyncio
async def test_validation_error(async_client):
    response = await async_client.post("/users", json={"username": "ab"})  # too short
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any("username" in str(e) for e in errors)

# pytest.ini
[pytest]
asyncio_mode = auto  # or use @pytest.mark.asyncio on each test

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import csv
import io

app = FastAPI()

# Stream a large CSV without loading it all into memory
@app.get("/export/users")
async def export_users():
    async def generate_csv():
        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerow(["id", "username", "email"])  # header
        yield output.getvalue()
        output.seek(0)
        output.truncate(0)

        for i in range(1_000_000):  # stream 1M rows
            writer.writerow([i, f"user{i}", f"user{i}@example.com"])
            if i % 1000 == 0:  # flush every 1000 rows
                yield output.getvalue()
                output.seek(0)
                output.truncate(0)
                await asyncio.sleep(0)  # yield control to event loop

    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"},
    )

# Server-Sent Events (SSE) for real-time updates
@app.get("/events")
async def event_stream():
    async def generate():
        for i in range(10):
            yield f"data: Event {i}\n\n"  # SSE format
            await asyncio.sleep(1)

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

# Stream a file from disk
@app.get("/download/{filename}")
async def download_file(filename: str):
    def iterfile():
        with open(f"/data/{filename}", "rb") as f:
            yield from f  # yields chunks as they are read

    return StreamingResponse(iterfile(), media_type="application/octet-stream")

# pip install strawberry-graphql[fastapi]
import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI
from typing import Optional

# Define types with Strawberry (type hints = schema)
@strawberry.type
class User:
    id:       int
    username: str
    email:    str

fake_users = [
    User(id=1, username="alice", email="alice@example.com"),
    User(id=2, username="bob",   email="bob@example.com"),
]

# Query type
@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> list[User]:
        return fake_users

    @strawberry.field
    def user(self, id: int) -> Optional[User]:
        return next((u for u in fake_users if u.id == id), None)

# Mutation type
@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, username: str, email: str) -> User:
        new_user = User(id=len(fake_users)+1, username=username, email=email)
        fake_users.append(new_user)
        return new_user

schema = strawberry.Schema(query=Query, mutation=Mutation)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

# REST routes coexist happily
@app.get("/health")
def health():
    return {"status": "ok"}

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI()

# Default 422 response body shape from FastAPI:
# {
#   "detail": [
#     {
#       "type": "missing",
#       "loc": ["body", "price"],
#       "msg": "Field required",
#       "input": {"name": "Widget"},
#       "url": "https://errors.pydantic.dev/..."
#     }
#   ]
# }

# Custom error handler — reshape the response
@app.exception_handler(RequestValidationError)
async def custom_validation_handler(
    request: Request, exc: RequestValidationError
) -> JSONResponse:
    errors = []
    for error in exc.errors():
        field = " -> ".join(str(loc) for loc in error["loc"])
        errors.append({"field": field, "message": error["msg"]})

    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "errors": errors,
        },
    )

class Product(BaseModel):
    name:  str = Field(min_length=1)
    price: float = Field(gt=0)
    stock: int   = Field(ge=0)

@app.post("/products")
def create_product(product: Product):
    return {"success": True, "product": product.model_dump()}

# Request: POST /products {}
# Custom response:
# {
#   "success": false,
#   "errors": [
#     {"field": "body -> name",  "message": "Field required"},
#     {"field": "body -> price", "message": "Field required"},
#     {"field": "body -> stock", "message": "Field required"}
#   ]
# }

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from typing import Annotated
import asyncpg

# Module-level: shared across all requests
db_pool: asyncpg.Pool | None = None
ml_model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global db_pool, ml_model

    # Create once at startup — shared by all workers in same process
    db_pool = await asyncpg.create_pool("postgresql://localhost/db", min_size=5)
    # ml_model = load_model("model.pkl")  # load once

    yield  # requests are served here

    await db_pool.close()

app = FastAPI(lifespan=lifespan)

# Request-scoped dependency — new connection per request from the pool
async def get_conn():
    async with db_pool.acquire() as conn:  # borrows from the shared pool
        yield conn                          # returned to pool after request

Conn = Annotated[asyncpg.Connection, Depends(get_conn)]

@app.get("/users")
async def list_users(conn: Conn):
    rows = await conn.fetch("SELECT id, username FROM users")
    return [dict(r) for r in rows]

# Singleton dependency — same instance for all requests (use_cache=True, default)
class AppState:
    def __init__(self):
        self.request_count = 0  # not thread-safe — example only

app_state = AppState()

def get_app_state() -> AppState:
    return app_state  # same object every request

@app.get("/stats")
def stats(state: Annotated[AppState, Depends(get_app_state)]):
    state.request_count += 1
    return {"total_requests": state.request_count}

from fastapi import FastAPI, Header, Cookie, Response
from typing import Annotated

app = FastAPI()

# Read request headers
@app.get("/headers")
def read_headers(
    user_agent: Annotated[str | None, Header()] = None,
    accept_language: Annotated[str | None, Header()] = None,
    x_api_key: Annotated[str | None, Header()] = None,  # X-Api-Key header
):
    # FastAPI converts Accept-Language → accept_language automatically
    return {
        "user_agent":      user_agent,
        "accept_language": accept_language,
        "x_api_key":       x_api_key,
    }

# Duplicate headers → list
@app.get("/multi-header")
def multi_header(
    x_token: Annotated[list[str] | None, Header()] = None,
):
    return {"x_token": x_token}  # all values of X-Token header

# Read cookies
@app.get("/profile")
def read_profile(
    session_id: Annotated[str | None, Cookie()] = None,
):
    return {"session_id": session_id}

# Set a cookie in the response
@app.post("/login")
def login(response: Response, username: str):
    response.set_cookie(
        key="session_id",
        value=f"session_{username}",
        httponly=True,   # not accessible via JS
        secure=True,     # HTTPS only
        samesite="lax",  # CSRF protection
        max_age=3600,    # 1 hour
    )
    return {"message": "Logged in"}

# Delete a cookie
@app.post("/logout")
def logout(response: Response):
    response.delete_cookie("session_id")
    return {"message": "Logged out"}