Python logoPythonINTERMEDIATE

FastAPI

Essential FastAPI reference covering routes, request validation, Pydantic models, dependencies, authentication, file uploads, middleware, database integration, and project structure.

10 min read
fastapipythonapiasyncrestweb

Sign in to mark items as known and track your progress.

Sign in

Installation & Setup

Install FastAPI and run your first application.

Quick Start

Install FastAPI with uvicorn and create a basic application.

python
# 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.py
💡 "fastapi[standard]" includes uvicorn, email-validator, and other essentials
⚡ Auto-generated docs at /docs (Swagger) and /redoc — no setup needed
📌 Use "fastapi dev" for development (auto-reload) and "fastapi run" for production
🟢 Both sync (def) and async (async def) route handlers are supported
installsetupquickstart

Path & Query Parameters

Define URL path parameters and query string parameters with validation.

Path Parameters

Extract values from the URL path with type validation.

python
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}
💡 Path parameters are required — they're part of the URL itself
⚡ Use Path() for validation: gt, ge, lt, le for numbers; min_length, max_length for strings
📌 Enum parameters auto-generate a dropdown in the /docs UI
🟢 Use {param:path} to capture a full file path including slashes
pathparametersvalidation

Query Parameters

Accept query string parameters with defaults and validation.

python
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}
💡 Any function parameter not in the URL path is automatically a query parameter
⚡ Use Query() for string validation: min_length, max_length, pattern (regex)
📌 Use list[str] with Query() to accept repeated query params (?tag=a&tag=b)
🟢 Set deprecated=True to mark a parameter as deprecated in the docs
queryparametersvalidation

Request Body & Pydantic Models

Define and validate request bodies using Pydantic models.

Pydantic Models

Define request and response schemas with automatic validation.

python
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 item
💡 Pydantic model parameters are automatically parsed from the JSON request body
⚡ Use Field() for validation: gt, ge, lt, le (numbers), min_length, max_length (strings)
📌 Nested models are fully validated — define Address as a model and use it inside User
🟢 FastAPI auto-detects: path params from URL, body from Pydantic models, rest as query
pydanticmodelsbodyvalidation

Response Models & Status Codes

Control response shape and HTTP status codes.

python
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 item
💡 response_model filters the output — use it to hide internal fields like passwords
⚡ You can use return type annotations (-> ItemOut) instead of response_model=
📌 status.HTTP_201_CREATED is clearer than the magic number 201
🟢 response_model_exclude_unset=True omits fields that weren't explicitly set
responsestatus-codesmodels

Form Data & File Uploads

Handle HTML form submissions and file uploads.

Form Data & File Uploads

Receive form fields and uploaded files in endpoints.

python
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)}
💡 Use Form() for form fields and File()/UploadFile for file uploads — not Pydantic models
⚡ UploadFile streams to disk — use it for large files; bytes = File() loads into memory
📌 You cannot mix JSON body (Pydantic) with form/file data in the same endpoint
🟢 Install python-multipart for form/file support: pip install python-multipart
formsfilesupload

Cookie & Header Parameters

Read cookies and HTTP headers from incoming requests.

Cookies & Headers

Extract cookie values and HTTP headers from requests.

python
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}
💡 Header names are auto-converted: X-Request-Id becomes x_request_id in Python
⚡ Use Response.set_cookie() to send cookies back to the client
📌 Always set httponly=True on session cookies to prevent XSS access
🟢 Use list[str] for headers that can appear multiple times (e.g., X-Token)
cookiesheadersparameters

Dependencies

Dependency injection for shared logic, DB sessions, and auth.

Dependency Injection

Share logic across endpoints using Depends().

python
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()
💡 Dependencies with yield run cleanup code after the response — perfect for DB sessions
⚡ Sub-dependencies chain automatically: get_admin_user → get_current_user → oauth2_scheme
📌 Use Depends() without arguments on a class to use it directly as a dependency
🟢 Global dependencies apply to every route — great for API key verification
dependenciesdependsinjection

Security & Authentication

Implement OAuth2 with JWT tokens and password hashing.

OAuth2 with JWT

Implement token-based authentication with password hashing.

python
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"}
💡 OAuth2PasswordBearer auto-adds a login button to the /docs UI
⚡ Install dependencies: pip install python-jose[cryptography] passlib[bcrypt]
📌 Never store plain-text passwords — always hash with bcrypt via passlib
🟢 Chain dependencies: protected routes just add user = Depends(get_current_user)
authjwtoauth2security

Error Handling

Raise HTTP errors and define custom exception handlers.

HTTPException & Custom Handlers

Return proper error responses and handle exceptions globally.

python
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]
💡 HTTPException is the standard way to return error responses in FastAPI
⚡ Use @app.exception_handler() to catch custom exceptions globally
📌 Override RequestValidationError handler to customize 422 validation responses
🟢 You can add custom headers to HTTPException for debugging or auth flows
errorsexceptionshttp

Middleware & CORS

Add middleware for cross-cutting concerns and configure CORS.

CORS & Custom Middleware

Configure CORS and write custom middleware for logging, timing, etc.

python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)
💡 Middleware runs on every request/response — use for logging, timing, auth checks
⚡ Set allow_origins to specific domains in production — never use ["*"] with credentials
📌 Middleware order matters — they execute in reverse order of how they're added
🟢 Built-in middleware: CORSMiddleware, TrustedHostMiddleware, HTTPSRedirectMiddleware
corsmiddleware

Database Integration

Connect to databases with SQLAlchemy or SQLModel.

SQLAlchemy Setup

Configure SQLAlchemy with FastAPI using dependency injection.

python
# 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):
    pass
💡 Use the yield dependency pattern to ensure DB sessions are always closed
⚡ For PostgreSQL, install: pip install psycopg2-binary
📌 check_same_thread is only needed for SQLite — remove it for other databases
🟢 Call Base.metadata.create_all() once at startup to create tables
databasesqlalchemysetup

CRUD Operations

Create, read, update, and delete records with SQLAlchemy.

python
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_item
💡 Use model_dump(exclude_unset=True) for PATCH-style partial updates
⚡ db.refresh() reloads the object from the DB — gets auto-generated fields like id
📌 Always check if the record exists before updating/deleting — raise 404 if not
🟢 Use sync def (not async def) for SQLAlchemy — it uses blocking I/O by default
cruddatabasesqlalchemy

Project Structure

Organize larger applications with APIRouter and multiple files.

APIRouter & Multiple Files

Split routes into separate files using APIRouter.

python
# 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)
💡 APIRouter works exactly like FastAPI() — same decorators, same parameters
⚡ Use tags=["items"] to group endpoints under a section in the docs UI
📌 Set prefix on the router to avoid repeating "/items" on every route
🟢 Add dependencies at include_router() level to protect entire route groups
routerstructureorganization

Advanced Features

Background tasks, lifespan events, streaming, WebSockets, and testing.

Background Tasks & Lifespan

Run tasks after returning a response and manage app startup/shutdown.

python
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"}
💡 Background tasks run after the response — the client doesn't wait for them
⚡ Use lifespan for startup tasks: loading ML models, connecting to DBs, etc.
📌 For heavy background work, use Celery or ARQ instead of BackgroundTasks
🟢 Store shared resources on app.state inside the lifespan context
backgroundlifespanstartupshutdown

Streaming, WebSockets & Testing

Stream responses, use WebSockets, and test with TestClient.

python
# 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()
💡 StreamingResponse with text/event-stream gives you Server-Sent Events (SSE)
⚡ TestClient uses requests-style API — no need to start a real server
📌 Use app.dependency_overrides to swap real DB for test DB in tests
🟢 WebSocket endpoints use await ws.receive_text() and ws.send_text() in a loop
streamingwebsockettestingsse