Python logoPythonv0.139INTERMEDIATE

FastAPI

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

10 min read
fastapipythonapiasyncrestweb
Loading your progress

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

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

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.

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.

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.

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.

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.

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.

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.

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.

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

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