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.
The documented default is now Annotated[type, Query(...)] — metadata in the type, the Python default (= None) stays as the default. The old = Query(None) style still works but is legacy.
Annotated[int, Path(ge=1)]
Required path int, validated ≥1.
Annotated[str | None, Query(min_length=3)] = None★
Optional query with min-length validation.
Annotated[list[str], Query()] = []
Accepts repeated: ?q=a&q=b
from typing import Annotated
from fastapi import Path, Query
@app.get("/users/{user_id}")
def get_user(
user_id: Annotated[int, Path(ge=1)], # required int ≥1
active: bool = True, # optional query
q: Annotated[str | None, Query(min_length=3)] = None,
):
return {"user_id": user_id, "active": active}
# legacy default-value style (still valid): q: str = Query(None)
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 typing import Annotated
from fastapi import Form, File, UploadFile, Cookie, Header
@app.post("/login/")
def login(
username: Annotated[str, Form()],
password: Annotated[str, Form()],
):
return {"user": username}
@app.post("/upload/")
async def upload(file: Annotated[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).
Wrap a dependency in an Annotated type-alias (DbSession = Annotated[Session, Depends(get_db)]) and reuse it across routes — declare the wiring once.
from typing import Annotated
from fastapi import Depends
def pagination(skip: int = 0, limit: int = 20):
return {"skip": skip, "limit": limit}
# reusable dependency alias — declare once, reuse everywhere
Pagination = Annotated[dict, Depends(pagination)]
@app.get("/items/")
def list_items(pg: Pagination):
return pg
# Yield dependency — cleanup always runs
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
DbSession = Annotated[Session, Depends(get_db)]
@app.get("/users/")
def get_users(db: DbSession):
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 typing import Annotated
from fastapi import Depends
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: Annotated[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}