01Install & Runinfra
One command installs everything; auto-docs at /docs and /redoc with zero config.
pip install "fastapi[standard]"★
Includes uvicorn, email-validator, and extras.
uvicorn main:app --reload★
Dev server with auto-reload. fastapi dev main.py also works.
fastapi run main.py
Production mode (no reload).
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
return {"message": "Hello FastAPI"}
02Route Decoratorsroutes
One decorator per HTTP verb. FastAPI auto-generates Swagger docs from type hints.
GET @app.get("/path")★
Read / fetch data. Returns JSON by default.
POST @app.post("/path")★
Create resource. Body is parsed from JSON.
PUT @app.put("/path/{id}")
Full replacement update.
PATCH @app.patch("/path/{id}")
Partial update.
DELETE @app.delete("/path/{id}")
Remove resource.
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"id": item_id}
@app.post("/items/", status_code=201)
def create_item(item: Item):
return item
@app.delete("/items/{id}")
def delete_item(id: int):
return {"deleted": id}
03Path & Query Parametersparams
Path params are in the URL. Everything else with a default is a query param. Use Path() / Query() to add validation.
Path(..., ge=1)
Validates ≥1; ... means required.
Query(None, min_length=3)★
Optional query with min-length validation.
q: list[str] = Query([])
Accepts repeated: ?q=a&q=b
from fastapi import Path, Query
@app.get("/users/{user_id}")
def get_user(
user_id: int = Path(..., ge=1), # required int ≥1
active: bool = True, # optional query
q: str = Query(None, min_length=3),
):
return {"user_id": user_id, "active": active}
04Request Body (Pydantic)pydantic
Declare a class inheriting BaseModel — FastAPI auto-validates, serializes, and documents the JSON body.
Pydantic v2 is the default since FastAPI 0.100+. Use model_config = ConfigDict(...) instead of inner Config class.
name: str★
Required field.
desc: str | None = None★
Optional field.
price: float = Field(..., gt=0)
Required with validation — must be >0.
Nested models
Just annotate a field with another BaseModel type.
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str
description: str | None = None
price: float = Field(..., gt=0)
tax: float = 0.0
@app.post("/items/")
def create_item(item: Item):
total = item.price * (1 + item.tax)
return {"total": total, **item.model_dump()}
05Response Models & Status Codespydantic
response_model filters output — never leaks password hashes or internal fields. Pair with status codes for correct REST semantics.
Use status.HTTP_201_CREATED constants over magic numbers.
from fastapi import status
from pydantic import BaseModel
class UserIn(BaseModel):
username: str; password: str
class UserOut(BaseModel):
username: str # no password here!
@app.post(
"/users/",
response_model=UserOut, # strips extra fields
status_code=status.HTTP_201_CREATED,
)
def create_user(user: UserIn) -> UserOut:
return user # password is filtered out
06Form · File · Cookie · Headerparams
Non-JSON parameter sources. Install python-multipart for Form/File support.
⚠ Cannot mix JSON Body and Form/File in the same endpoint.
Form(...)
HTML form field. Needs python-multipart.
UploadFile = File(...)★
Streams to disk. Better than bytes for large files.
Cookie(None)
Read HTTP cookies. None → optional.
Header(None)
FastAPI auto-converts user_agent → User-Agent.
from fastapi import Form, File, UploadFile, Cookie, Header
@app.post("/login/")
def login(
username: str = Form(...),
password: str = Form(...),
):
return {"user": username}
@app.post("/upload/")
async def upload(file: UploadFile = File(...)):
contents = await file.read()
return {"name": file.filename, "size": len(contents)}
07Dependency Injectionpydantic / DI
Share logic (auth, DB sessions, pagination) across routes with Depends(). Dependencies can nest and chain automatically.
Use yield for cleanup (e.g., close DB session after response).
from fastapi import Depends
def pagination(skip: int = 0, limit: int = 20):
return {"skip": skip, "limit": limit}
@app.get("/items/")
def list_items(pg: dict = Depends(pagination)):
return pg
# Yield dependency — cleanup always runs
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
def get_users(db = Depends(get_db)):
return db.query(User).all()
08Security & OAuth2 / JWTsecurity
OAuth2PasswordBearer adds a login button to /docs. Combine with python-jose for JWT and passlib for bcrypt hashing.
⚠ Never store plain-text passwords. Always hash with bcrypt via passlib.
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
oauth2 = OAuth2PasswordBearer(tokenUrl="token")
pwd_ctx = CryptContext(schemes=["bcrypt"])
def get_hash(pw): return pwd_ctx.hash(pw)
def verify(pw, hashed): return pwd_ctx.verify(pw, hashed)
@app.get("/users/me")
def read_me(token: str = Depends(oauth2)):
user = decode_jwt(token) # verify via python-jose
return user
# API Key alternative
from fastapi.security import APIKeyHeader
api_key = APIKeyHeader(name="X-API-Key")
09HTTP Exceptions & Error Handlingsecurity
Always use HTTPException — never return a raw error dict. Register custom handlers for global behavior.
Override RequestValidationError to customize the 422 response format.
raise HTTPException(404, detail="...")★
Abort with an HTTP error code.
headers={"WWW-Authenticate":"Bearer"}
Custom headers on the error response.
@app.exception_handler(MyError)
Catch custom exception types globally.
from fastapi import HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@app.get("/items/{id}")
def read_item(id: int):
if id not in items:
raise HTTPException(
status_code=404, detail="Item not found"
)
@app.exception_handler(RequestValidationError)
async def val_handler(req, exc):
return JSONResponse(
{"errors": exc.errors()}, status_code=422
)
10Middleware & CORSinfra
Middleware wraps every request/response. CORS is the most common middleware for browser-facing APIs.
⚠ Never use allow_origins=["*"] with allow_credentials=True in production.
from fastapi.middleware.cors import CORSMiddleware
from fastapi import Request
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["*"], allow_headers=["*"],
)
# Custom middleware — timing / logging
@app.middleware("http")
async def add_timing(request: Request, call_next):
import time; t = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - t)
return response
11APIRouter & Project Structureinfra
Split large apps into multiple files. APIRouter works exactly like FastAPI().
Use prefix= to avoid repeating /users on every route.
# routers/users.py
from fastapi import APIRouter
router = APIRouter()
@router.get("/", tags=["users"])
def list_users(): return []
@router.get("/{id}")
def get_user(id: int): ...
# main.py
from routers import users
app.include_router(
users.router,
prefix="/users",
dependencies=[Depends(verify_token)], # protects all routes
)
12Background Tasks & Lifespaninfra
Background tasks run after the response is sent. Lifespan loads/tears down resources (ML models, DB pools) at startup/shutdown.
For heavy CPU/long work use Celery or ARQ — not BackgroundTasks.
from fastapi import BackgroundTasks
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# startup — load model once, not per request
app.state.model = load_ml_model()
yield
# shutdown — release resources
app.state.model = None
app = FastAPI(lifespan=lifespan)
def send_email(to: str, body: str):
... # slow I/O, runs after response
@app.post("/notify/")
def notify(email: str, bg: BackgroundTasks):
bg.add_task(send_email, email, "Welcome!")
return {"status": "queued"}
13WebSockets & Streaminginfra
WebSockets enable real-time bidirectional comms. StreamingResponse with text/event-stream gives Server-Sent Events (SSE — great for LLM token streaming).
from fastapi import WebSocket
from fastapi.responses import StreamingResponse
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await ws.accept()
while True:
data = await ws.receive_text()
await ws.send_text(f"echo: {data}")
# SSE — LLM token streaming
def token_gen(prompt: str):
for token in llm.stream(prompt):
yield f"data: {token}\n\n"
@app.get("/stream")
def stream(prompt: str):
return StreamingResponse(
token_gen(prompt), media_type="text/event-stream"
)
14Database & Testinginfra
SQLAlchemy via yield-dependency ensures sessions always close. TestClient needs no real server.
Use app.dependency_overrides to swap the real DB for a test DB.
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("sqlite:///./test.db")
SessionLocal = sessionmaker(bind=engine, autoflush=False)
# testing
from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_root():
r = client.get("/")
assert r.status_code == 200
assert r.json() == {"message": "Hello FastAPI"}
# Override DB for tests
app.dependency_overrides[get_db] = lambda: test_db_session
⚡ML Model Inference EndpointAI systems
Load the model once in lifespan, not inside the handler — avoids re-loading on every request.
UploadFile streams the file; use await file.read() for small images, chunked reading for large payloads.
from contextlib import asynccontextmanager
import numpy as np
@asynccontextmanager
async def lifespan(app):
app.state.model = load_model("model.pkl")
yield
app = FastAPI(lifespan=lifespan)
class PredictRequest(BaseModel):
features: list[float]
@app.post("/predict", response_model=dict)
def predict(req: PredictRequest, request: Request):
model = request.app.state.model
X = np.array([req.features])
pred = model.predict(X).tolist()
return {"prediction": pred}