FastAPI
Essential FastAPI reference covering routes, request validation, Pydantic models, dependencies, authentication, file uploads, middleware, database integration, and project structure.
Other Python Sheets
Sign in to mark items as known and track your progress.
Sign inInstallation & Setup
Install FastAPI and run your first application.
Quick Start
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.
Path Parameters
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}Query Parameters
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.
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 itemResponse Models & Status Codes
Control 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.
Form Data & 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.
Cookies & Headers
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.
Dependency Injection
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.
OAuth2 with JWT
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.
HTTPException & Custom 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.
CORS & Custom Middleware
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.
SQLAlchemy Setup
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):
passCRUD Operations
Create, 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.
APIRouter & 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.
Background Tasks & Lifespan
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"}Streaming, WebSockets & Testing
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()