FastAPI
Essential FastAPI reference covering routes, request validation, Pydantic models, dependencies, authentication, file uploads, middleware, database integration, and project structure.
Other Python Sheets
Installation & Setup
Install FastAPI and run your first application.
Install FastAPI with uvicorn and create a basic application.
# Install
pip install "fastapi[standard]"
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
# Run
fastapi dev main.pyPath & Query Parameters
Define URL path parameters and query string parameters with validation.
Extract values from the URL path with type validation.
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
# With validation
@app.get("/users/{user_id}")
async def read_user(
user_id: int = Path(gt=0, le=1000, description="The user ID"),
):
return {"user_id": user_id}Accept query string parameters with defaults and validation.
from fastapi import FastAPI, Query
app = FastAPI()
# Basic query parameters
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# GET /items/?skip=5&limit=20
# Required + validated
@app.get("/search/")
async def search(
q: str = Query(min_length=3, max_length=50),
):
return {"query": q}Request Body & Pydantic Models
Define and validate request bodies using Pydantic models.
Define request and response schemas with automatic validation.
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str
price: float = Field(gt=0, description="Must be positive")
description: str | None = None
tags: list[str] = []
@app.post("/items/")
async def create_item(item: Item):
return itemControl response shape and HTTP status codes.
from fastapi import FastAPI, status
class ItemOut(BaseModel):
name: str
price: float
# password field excluded from response
@app.post("/items/", response_model=ItemOut, status_code=201)
async def create_item(item: Item):
return itemForm Data & File Uploads
Handle HTML form submissions and file uploads.
Receive form fields and uploaded files in endpoints.
from fastapi import FastAPI, File, Form, UploadFile
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
return {"username": username}
@app.post("/upload/")
async def upload(file: UploadFile):
contents = await file.read()
return {"filename": file.filename, "size": len(contents)}Cookie & Header Parameters
Read cookies and HTTP headers from incoming requests.
Extract cookie values and HTTP headers from requests.
from fastapi import Cookie, Header
@app.get("/items/")
async def read_items(
session_id: str | None = Cookie(default=None),
user_agent: str | None = Header(default=None),
):
return {"session": session_id, "agent": user_agent}Dependencies
Dependency injection for shared logic, DB sessions, and auth.
Share logic across endpoints using Depends().
from fastapi import Depends
async def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
return db.query(Item).all()Security & Authentication
Implement OAuth2 with JWT tokens and password hashing.
Implement token-based authentication with password hashing.
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
user = authenticate(form.username, form.password)
token = create_access_token({"sub": user.username})
return {"access_token": token, "token_type": "bearer"}Error Handling
Raise HTTP errors and define custom exception handlers.
Return proper error responses and handle exceptions globally.
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return items[item_id]Middleware & CORS
Add middleware for cross-cutting concerns and configure CORS.
Configure CORS and write custom middleware for logging, timing, etc.
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)Database Integration
Connect to databases with SQLAlchemy or SQLModel.
Configure SQLAlchemy with FastAPI using dependency injection.
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
engine = create_engine("sqlite:///./app.db")
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
passCreate, read, update, and delete records with SQLAlchemy.
from fastapi import Depends
from sqlalchemy.orm import Session
@app.post("/items/", response_model=ItemOut, status_code=201)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
db_item = Item(**item.model_dump())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_itemProject Structure
Organize larger applications with APIRouter and multiple files.
Split routes into separate files using APIRouter.
# app/routers/items.py
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["items"])
@router.get("/")
async def read_items():
return []
# app/main.py
from fastapi import FastAPI
from app.routers import items, users
app = FastAPI()
app.include_router(items.router)
app.include_router(users.router)Advanced Features
Background tasks, lifespan events, streaming, WebSockets, and testing.
Run tasks after returning a response and manage app startup/shutdown.
from fastapi import BackgroundTasks
@app.post("/notify/")
async def send_notification(
email: str, background_tasks: BackgroundTasks,
):
background_tasks.add_task(send_email, email)
return {"message": "Notification queued"}Stream responses, use WebSockets, and test with TestClient.
# Streaming
from fastapi.responses import StreamingResponse
@app.get("/stream")
async def stream():
async def generate():
for i in range(10):
yield f"chunk {i}\n"
return StreamingResponse(generate(), media_type="text/plain")
# WebSocket
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
await ws.accept()
data = await ws.receive_text()
await ws.send_text(f"Echo: {data}")
await ws.close()