Path Operations & Pydantic

App Setup & Routing

from fastapi import FastAPI, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(
    title="My API",
    description="FastAPI example",
    version="1.0.0",
    docs_url="/docs",       # Swagger UI
    redoc_url="/redoc",     # ReDoc
)

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

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

@app.get("/items/{item_id}")
def get_item(item_id: int):            # auto-validates int
    return {"id": item_id}

# Query parameters
@app.get("/users")
def list_users(
    page: int = 1,
    limit: int = 10,
    search: str | None = None,         # optional
    active: bool = True,
):
    return {"page": page, "limit": limit, "search": search}

# Multiple path params
@app.get("/users/{user_id}/posts/{post_id}")
def get_user_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

# Response status codes
@app.post("/users", status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    return {"id": 1, **user.model_dump()}

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int):
    pass

# HTTP errors
def get_or_404(item_id: int):
    item = db.get(item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found",
        )
    return item

Pydantic Models

from pydantic import BaseModel, Field, EmailStr, field_validator, model_validator
from typing import Optional
from datetime import datetime
from enum import Enum

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

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    role: UserRole = UserRole.user

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

    @field_validator("password")
    @classmethod
    def password_must_have_uppercase(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain uppercase letter")
        return v

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}  # enables ORM mode

class UserUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=2)
    bio: Optional[str] = None

# Nested models
class Address(BaseModel):
    street: str
    city: str
    country: str = "US"

class UserWithAddress(UserBase):
    address: Optional[Address] = None
    tags: list[str] = []
    metadata: dict[str, str] = {}

# Usage
user = UserCreate(email="user@example.com", name="Alice", password="Password1")
print(user.model_dump())
print(user.model_dump_json())
user_dict = {"email": "a@b.com", "name": "Bob", "password": "StrongPass1"}
user = UserCreate.model_validate(user_dict)

Request Body & Response Models

from fastapi import FastAPI, Body
from fastapi.responses import JSONResponse, StreamingResponse

# Request body
@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate) -> UserResponse:
    # response_model filters output — only UserResponse fields returned
    db_user = create_in_db(user)
    return db_user

# Multiple body params
@app.put("/users/{user_id}")
def update_user(
    user_id: int,
    user: UserUpdate,
    reason: str = Body(..., embed=True),  # extra body field
):
    return {"id": user_id, "reason": reason, **user.model_dump(exclude_none=True)}

# response_model_exclude / include
@app.get("/users/me", response_model=UserResponse, response_model_exclude={"password"})
def get_me():
    ...

# List response
@app.get("/users", response_model=list[UserResponse])
def list_users():
    return db.get_all_users()

# Custom response
@app.get("/export")
def export():
    def generate():
        yield "col1,col2
"
        yield "a,b
"
    return StreamingResponse(generate(), media_type="text/csv",
                             headers={"Content-Disposition": "attachment; filename=data.csv"})

Routers (APIRouter)

# routers/users.py
from fastapi import APIRouter

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

@router.get("/")
def list_users():
    return []

@router.get("/{user_id}")
def get_user(user_id: int):
    return {"id": user_id}

# main.py
from routers import users, posts, auth

app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router)
app.include_router(posts.router, prefix="/api/v1")

Dependency Injection & Async

Depends() — Dependency Injection

from fastapi import Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

# Simple dependency
def common_pagination(page: int = 1, limit: int = 10):
    return {"skip": (page - 1) * limit, "limit": limit}

@app.get("/items")
def list_items(pagination: dict = Depends(common_pagination)):
    return {"pagination": pagination}

# Auth dependency
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = await db.get_user(user_id)
    if user is None:
        raise credentials_exception
    return user

# Use auth dependency
@app.get("/profile")
async def get_profile(current_user: User = Depends(get_current_user)):
    return current_user

# Database session dependency
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

Async Route Handlers

import asyncio
import httpx
from fastapi import BackgroundTasks

# Async route — use for I/O bound work (DB, HTTP, files)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await db.fetch_user(user_id)   # awaitable DB call
    return user

# Sync route — use for CPU bound work; runs in threadpool
@app.get("/process")
def process_data():
    result = heavy_cpu_task()            # runs in threadpool automatically
    return result

# Background tasks
@app.post("/email")
async def send_email(
    email: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(send_welcome_email, email)  # runs after response
    return {"message": "Email queued"}

# Parallel async calls
@app.get("/dashboard")
async def dashboard(user_id: int):
    users_task = asyncio.create_task(db.get_user(user_id))
    orders_task = asyncio.create_task(db.get_orders(user_id))
    stats_task = asyncio.create_task(db.get_stats(user_id))
    user, orders, stats = await asyncio.gather(users_task, orders_task, stats_task)
    return {"user": user, "orders": orders, "stats": stats}

# App lifespan — startup/shutdown
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await db.connect()
    await redis.connect()
    yield
    # Shutdown
    await db.disconnect()
    await redis.disconnect()

app = FastAPI(lifespan=lifespan)

SQLAlchemy Async + Alembic

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=False, pool_size=10, max_overflow=20)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# models.py
from sqlalchemy import String, Boolean, Integer, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    name: Mapped[str] = mapped_column(String)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    orders: Mapped[list["Order"]] = relationship(back_populates="user")

# CRUD with async SQLAlchemy
from sqlalchemy import select, update, delete

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100):
    result = await db.execute(select(User).offset(skip).limit(limit))
    return result.scalars().all()

async def create_user(db: AsyncSession, email: str, name: str) -> User:
    user = User(email=email, name=name)
    db.add(user)
    await db.flush()   # get ID without committing
    return user

Middleware & Exception Handlers

from fastapi import Request
from fastapi.responses import JSONResponse
import time

# Middleware
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

# Custom exception
class AppError(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message},
    )

FastAPI Interview Questions

Q: What makes FastAPI fast?

Two things: (1) It is built on Starlette (ASGI) and uses async I/O — async route handlers don't block threads on I/O waits. (2) Pydantic v2 uses Rust-based validation (pydantic-core) making serialization/validation extremely fast. FastAPI also generates OpenAPI docs at zero runtime cost (metadata only).

Q: How does FastAPI handle request validation?

FastAPI uses Python type hints and Pydantic models to automatically validate request data. Path parameters, query params, and request body are all validated before the route handler runs. Invalid data returns a 422 Unprocessable Entity with a detailed error list. No manual validation code is needed.

Q: What is the Depends() system?

Depends() is FastAPI's dependency injection system. A dependency is any callable (function or class) whose return value is injected into the route handler. Dependencies can be nested (a dependency can depend on other dependencies), and FastAPI handles their resolution automatically. Common uses: database sessions, auth, pagination parameters, feature flags.

Q: async def vs def in FastAPI routes?

Use async def when the route does I/O (async database calls, HTTP requests, file reads). FastAPI awaits it directly on the event loop. Use def for CPU-bound work — FastAPI automatically runs sync routes in a threadpool executor so they don't block the event loop. Mixing async and sync incorrectly (e.g., calling blocking code in async def) will stall the server.

Q: What is response_model used for?

response_model specifies the Pydantic model for the response, which (1) filters out fields not in the model (e.g., passwords), (2) validates response data, and (3) generates the correct OpenAPI schema for docs. This keeps internal implementation details out of API responses without extra serialization logic.

Q: How do you run FastAPI in production?

# Development
uvicorn main:app --reload --port 8000

# Production — use Gunicorn with Uvicorn workers
gunicorn main:app -k uvicorn.workers.UvicornWorker --workers 4 --bind 0.0.0.0:8000

# Or with uvicorn directly (single process, use multiple containers for scale)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1

Q: How do you test FastAPI endpoints?

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={"email": "a@b.com", "name": "Alice", "password": "Strong1"})
    assert response.status_code == 201
    assert response.json()["email"] == "a@b.com"

def test_get_user_not_found():
    response = client.get("/users/99999")
    assert response.status_code == 404

# Override dependencies in tests
from main import app, get_db

def override_get_db():
    yield test_db_session

app.dependency_overrides[get_db] = override_get_db