Fastapi

FastAPI

This skill provides comprehensive guidance for building APIs with FastAPI.

Quick Navigation

When to Use

• Creating REST APIs with Python
• Adding endpoints with automatic validation
• Implementing OAuth2/JWT authentication
• Working with Pydantic models
• Adding dependency injection
• Configuring CORS, middleware
• Uploading files, handling forms
• Testing API endpoints

Installation

Requires Python 3.10+. Install: pip install "fastapi[standard]" (full with uvicorn) or pip install fastapi (minimal). Add python-multipart for forms/files.

Release Highlights (0.133.0 → 0.136.1)

• 0.134.0: streaming JSON Lines and streaming binary data support using yield.
• 0.135.0: first-class Server-Sent Events (SSE) support (EventSourceResponse).
• 0.135.1: fix around TaskGroup usage in request async exit stack (stability fix).
• 0.136.1: FastAPI updates its Pydantic v2 code to avoid deprecations and bumps Starlette to 1.0.0.

Patch Notes (0.136.2 → 0.136.3)

• SSE responses now validate event fields more strictly, so malformed ServerSentEvent payloads fail earlier instead of quietly streaming invalid frames.
• Header parameters no longer accept underscore-named incoming headers when convert_underscores=True (the default). If a client truly sends underscore headers, declare Header(convert_underscores=False) and verify that your proxy chain allows them.

Quick Start

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

Run: fastapi dev main.py

Core Patterns

Type-Safe Parameters

from typing import Annotated
from fastapi import Path, Query

@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(ge=1)],
    q: Annotated[str | None, Query(max_length=50)] = None
):
    return {"item_id": item_id, "q": q}

Request Body with Validation

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)

@app.post("/items/", response_model=Item)
def create_item(item: Item):
    return item

Dependencies

from fastapi import Depends

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def list_users(db: Annotated[Session, Depends(get_db)]):
    return db.query(User).all()

Authentication

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    return decode_token(token)

@app.get("/users/me")
def read_me(user: Annotated[User, Depends(get_current_user)]):
    return user

API Documentation

• Swagger UI: /docs
• ReDoc: /redoc
• OpenAPI: /openapi.json

Best Practices

• Use Annotated[Type, ...] for parameters
• Define Pydantic models for request/response
• Use response_model for output filtering
• Add status_code for proper HTTP codes
• Use tags for API organization
• Add dependencies at router/app level for auth

Prohibitions

• ❌ Return raw database models (use response models)
• ❌ Store passwords in plain text (use bcrypt/passlib)
• ❌ Mix Body with Form/File in same endpoint
• ❌ Use sync blocking I/O in async endpoints
• ❌ Skip HTTPException for error handling

Links

• Documentation
• Releases
• GitHub
• PyPI

Background Tasks

Run tasks after returning the response.

Use Cases

• Send email notifications
• Process uploaded files
• Update caches
• Log analytics

Basic Usage

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message: str = ""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Add Task Parameters

background_tasks.add_task(
    function,       # Task function
    arg1,           # Positional args
    arg2,
    key1=value1,    # Keyword args
    key2=value2,
)

Task Function Types

• Use def for I/O-bound tasks (file write, sync calls)
• Use async def for async operations (FastAPI handles both)

# Sync task
def process_file(path: str):
    with open(path) as f:
        # process...

# Async task
async def call_external_api(url: str):
    async with httpx.AsyncClient() as client:
        await client.get(url)

With Dependency Injection

from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q

@app.post("/send-notification/{email}")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
    q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

All tasks from dependencies and path operations are merged and run after response.

Heavy Background Jobs

For complex workloads, consider:

• Celery - Distributed task queue with RabbitMQ/Redis
• Redis Queue (RQ) - Simple Redis-based queue
• Dramatiq - Alternative to Celery

Use these when you need:

• Tasks running on multiple processes/servers
• Retries and failure handling
• Scheduled/periodic tasks
• Task result tracking

When to Use BackgroundTasks

✅ Good for:

• Small, quick tasks
• Access to same app variables/objects
• No need for separate worker process

❌ Better use Celery for:

• Long-running computations
• Tasks needing multiple workers/servers
• Complex retry logic

Import Note

# Use BackgroundTasks (plural) not BackgroundTask
from fastapi import BackgroundTasks  # ✓
# NOT from starlette.background import BackgroundTask

Bigger Applications - Multiple Files

Structure large FastAPI apps using APIRouter for modular organization.

Project Structure

app/
├── __init__.py
├── main.py              # Main FastAPI app
├── dependencies.py      # Shared dependencies
├── routers/
│   ├── __init__.py
│   ├── users.py        # User routes
│   └── items.py        # Item routes
└── internal/
    ├── __init__.py
    └── admin.py        # Admin routes (shared)

APIRouter

Create modular routers that work like mini FastAPI apps:

# routers/users.py
from fastapi import APIRouter

router = APIRouter()

@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

Router with Prefix and Dependencies

Apply common settings to all routes in a router:

# routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",               # All routes start with /items
    tags=["items"],               # OpenAPI tag
    dependencies=[Depends(get_token_header)],  # Auth for all routes
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def read_items():
    return fake_items_db

@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

Router Lifecycle Hooks (v0.128.6)

APIRouter supports startup and shutdown hooks (fixed in 0.128.6):

from fastapi import APIRouter

async def connect_db():
    ...

async def close_db():
    ...

router = APIRouter(
    on_startup=[connect_db],
    on_shutdown=[close_db],
)

Shared Dependencies

# dependencies.py
from typing import Annotated
from fastapi import Header, HTTPException

async def get_token_header(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No token provided")

Main App

Include routers in the main app:

# main.py
from fastapi import Depends, FastAPI
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])  # Global dependency

# Include routers
app.include_router(users.router)
app.include_router(items.router)

# Include with custom prefix/dependencies
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

Relative Imports

# From routers/items.py, import from parent package
from ..dependencies import get_token_header  # app/dependencies.py

• Single dot . = same package
• Double dots .. = parent package
• Triple dots ... = grandparent (rarely needed)

Include Router in Router

# Nest routers before including in main app
router.include_router(other_router)

Multiple Prefixes for Same Router

# Same router at different paths (e.g., /api/v1 and /api/latest)
app.include_router(api_router, prefix="/api/v1")
app.include_router(api_router, prefix="/api/latest")

Key Points

• APIRouter = mini FastAPI class with same parameters
• prefix must not end with /
• Router dependencies execute before decorator dependencies
• Path operations are "cloned" (not mounted) to include in OpenAPI schema
• Performance: including routers happens at startup (microseconds)

Body - Advanced Topics

Advanced request body handling patterns.

Multiple Body Parameters

Multiple Pydantic models = JSON keys by parameter names:

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    return {"item_id": item_id, "item": item, "user": user}

Expected body:

{
  "item": { "name": "Foo", "price": 42.0 },
  "user": { "username": "dave", "full_name": "Dave Grohl" }
}

Singular Values in Body

Use Body() for non-model values in body:

from typing import Annotated
from fastapi import Body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: Annotated[int, Body()]
):
    return {"item_id": item_id, "importance": importance}

Expected body:

{
    "item": {...},
    "user": {...},
    "importance": 5
}

Embed Single Body Parameter

Force key wrapping for single model:

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Annotated[Item, Body(embed=True)]
):
    return {"item_id": item_id, "item": item}

With embed=True:

{ "item": { "name": "Foo", "price": 42.0 } }

Without embed=True:

{ "name": "Foo", "price": 42.0 }

Field Validation

Use Field from Pydantic for model attributes:

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None,
        title="The description of the item",
        max_length=300
    )
    price: float = Field(gt=0, description="Must be greater than zero")
    tax: float | None = None

Field parameters: same as Query, Path, Body.

Nested Models

Use Pydantic models as types in other models:

from pydantic import BaseModel, HttpUrl

class Image(BaseModel):
    url: HttpUrl  # Validated URL
    name: str

class Item(BaseModel):
    name: str
    price: float
    images: list[Image] | None = None

Special Types

• set[str] - Unique items, deduplicated
• list[str] - Regular list
• dict[int, float] - Keys convert from JSON strings
• HttpUrl - Validated URL
• EmailStr - Validated email (requires pydantic[email])

Bodies of Pure Lists

@app.post("/images/multiple/")
async def create_multiple_images(images: list[Image]):
    return images

Arbitrary Dict Bodies

@app.post("/index-weights/")
async def create_index_weights(weights: dict[int, float]):
    return weights

JSON keys (strings) are converted to int.

Extra Data Types

from datetime import datetime, timedelta
from uuid import UUID

@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: Annotated[datetime, Body()],
    process_after: Annotated[timedelta, Body()],
):
    start_process = start_datetime + process_after
    return {"item_id": item_id, "start_process": start_process}

Cookie and Header Parameters

Handling HTTP cookies and headers in FastAPI.

Cookie Parameters

from typing import Annotated
from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}

Key Points

• Import: from fastapi import Cookie
• Required: Use Cookie() to distinguish from query params
• Validation: Same as Query(), Path()
• Browser limitation: Swagger UI cannot test cookies (JavaScript restriction)

Optional Cookie

@app.get("/items/")
async def read_items(
    session_id: Annotated[str | None, Cookie()] = None,
    tracking_id: Annotated[str | None, Cookie()] = None
):
    return {"session": session_id, "tracking": tracking_id}

Header Parameters

from typing import Annotated
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}

Automatic Conversion

FastAPI converts:

• user_agent → User-Agent
• x_token → X-Token

# Python variable: x_custom_header
# HTTP Header: X-Custom-Header (automatic conversion)
@app.get("/items/")
async def read_items(x_custom_header: Annotated[str | None, Header()] = None):
    return {"header": x_custom_header}

Disable Underscore Conversion

@app.get("/items/")
async def read_items(
    strange_header: Annotated[str | None, Header(convert_underscores=False)] = None
):
    return {"strange_header": strange_header}

⚠️ Warning: Some proxies/servers disallow headers with underscores.

0.136.3 note:

• With the default Header() behavior (convert_underscores=True), FastAPI now rejects incoming underscore-named headers instead of silently accepting them through the hyphen-conversion path.
• If a legacy client really sends X_Custom_Header, opt into Header(convert_underscores=False) explicitly and verify every proxy/load balancer in front of FastAPI preserves underscore headers.

Duplicate Headers

For headers that can appear multiple times:

@app.get("/items/")
async def read_items(x_token: Annotated[list[str] | None, Header()] = None):
    return {"X-Token values": x_token}

Request with:

X-Token: foo
X-Token: bar

Response:

{ "X-Token values": ["bar", "foo"] }

Common Headers

Authorization

@app.get("/protected/")
async def protected(authorization: Annotated[str | None, Header()] = None):
    if not authorization:
        raise HTTPException(401, "Missing authorization")
    # Parse "Bearer <token>"
    return {"auth": authorization}

Content-Type

@app.post("/webhook/")
async def webhook(
    content_type: Annotated[str | None, Header()] = None,
    body: bytes = Body(...)
):
    if content_type == "application/json":
        # Handle JSON
        pass
    return {"content_type": content_type}

Strict JSON Content-Type checking (FastAPI 0.132.0)

FastAPI 0.132.0 enables strict JSON `Content-Type` checking by default. If a request includes a JSON body but does not send a valid JSON Content-Type header (for example application/json), FastAPI will reject the request.

If you need the previous (more permissive) behavior for legacy clients, disable the check at app creation time:

from fastapi import FastAPI

app = FastAPI(strict_content_type=False)

Custom Headers

@app.get("/items/")
async def read_items(
    x_request_id: Annotated[str | None, Header()] = None,
    x_correlation_id: Annotated[str | None, Header()] = None
):
    return {"request_id": x_request_id, "correlation_id": x_correlation_id}

Cookie/Header Parameter Models (FastAPI 0.115+)

Cookie Model

from pydantic import BaseModel

class Cookies(BaseModel):
    session_id: str
    tracking_id: str | None = None

@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
    return cookies

Header Model

class CommonHeaders(BaseModel):
    x_request_id: str
    x_correlation_id: str | None = None
    user_agent: str | None = None

@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
    return headers

Setting Response Cookies/Headers

Set Cookie

from fastapi import Response

@app.post("/login/")
async def login(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=3600
    )
    return {"message": "logged in"}

Set Header

@app.get("/items/")
async def read_items(response: Response):
    response.headers["X-Custom-Header"] = "custom-value"
    return {"items": []}

Recipes

API Key in Header

from fastapi import Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != "secret-api-key":
        raise HTTPException(403, "Invalid API key")
    return api_key

@app.get("/protected/", dependencies=[Depends(verify_api_key)])
async def protected_route():
    return {"message": "Access granted"}

Request Tracing

import uuid

@app.middleware("http")
async def add_request_id(request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

Comparison Table

CORS (Cross-Origin Resource Sharing)

Enable cross-origin requests from frontend to backend.

What is Origin?

Origin = protocol + domain + port

Different origins:

• http://localhost (port 80)
• http://localhost:8080 (port 8080)
• https://localhost (HTTPS)

Basic CORS Setup

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost",
    "http://localhost:3000",
    "http://localhost:8080",
    "https://myapp.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Configuration Options

Common Configurations

Development (Allow All)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

⚠️ Can't use ["*"] with allow_credentials=True

Production (Specific Origins)

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://myapp.com",
        "https://www.myapp.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Regex Pattern

app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"https://.*\.myapp\.com",
)

How CORS Works

Preflight Requests

• Browser sends OPTIONS request first
• Includes Origin and Access-Control-Request-Method headers
• Backend responds with allowed origins/methods
• Browser then sends actual request

Simple Requests

• GET, HEAD, POST with standard headers
• No preflight needed
• Backend adds CORS headers to response

Credentials Warning

When allow_credentials=True:

• Cannot use ["*"] for origins, methods, or headers
• Must explicitly list all allowed values

# ❌ Invalid
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,  # Conflict!
)

# ✅ Valid
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
)

Debugging CORS

Check browser console for:

• Access-Control-Allow-Origin header missing
• Origin not in allowed list
• Method not allowed
• Credential issues

Use browser DevTools Network tab to inspect preflight requests.

````markdown

Dependencies

Dependency Injection system for shared logic, database connections, security.

Basic Dependency

from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

How It Works

1. Define a function (dependency) that takes parameters 2. Use Depends(function) in path operation parameters 3. FastAPI calls the dependency, gets result, passes to your function

Reusable Type Alias

CommonsDep = Annotated[dict, Depends(common_parameters)]

@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons

Classes as Dependencies

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    # Shortcut: Depends() without argument when class == type hint
    return {"q": commons.q, "skip": commons.skip}

Any "callable" works: function, class, or object with __call__.

Sub-dependencies

Dependencies can depend on other dependencies:

def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None
):
    return q or last_query

@app.get("/items/")
async def read_query(query: Annotated[str, Depends(query_or_cookie_extractor)]):
    return {"q": query}

use_cache Parameter

# Default: use_cache=True - same instance reused in request
dep1: Annotated[dict, Depends(get_db)]
dep2: Annotated[dict, Depends(get_db)]  # Same instance as dep1

# Force new instance
dep2: Annotated[dict, Depends(get_db, use_cache=False)]

Dependencies in Decorators

For side-effect dependencies (auth checks, logging) that don't return values:

async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="Invalid token")

@app.get("/items/", dependencies=[Depends(verify_token)])
async def read_items():
    return [{"item": "Foo"}]

Global Dependencies

Apply to all routes:

app = FastAPI(dependencies=[Depends(verify_token)])

Dependencies with yield

For setup/cleanup (e.g., database sessions):

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items/")
async def read_items(db: Annotated[Session, Depends(get_db)]):
    return db.query(Item).all()

Execution Order

1. Code before yield runs first 2. yield db passes value to endpoint 3. Endpoint executes 4. Code after yield runs (cleanup)

Exception Handling

async def get_db():
    db = DBSession()
    try:
        yield db
    except SomeException:
        db.rollback()
        raise
    finally:
        db.close()

⚠️ Cannot raise HTTPException AFTER yield - response already sent.

Sub-dependencies with yield

Exit order is reverse of entry (LIFO):

Enter dependency_a → Enter dependency_b → Endpoint → Exit dependency_b → Exit dependency_a

Recipes

Database Session per Request

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
def create_user(user: UserCreate, db: Annotated[Session, Depends(get_db)]):
    db_user = User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Authentication Chain

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user = decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.get("/users/me")
async def read_users_me(user: Annotated[User, Depends(get_current_active_user)]):
    return user

Parameterized Dependency

def pagination_params(max_limit: int = 100):
    def inner(skip: int = 0, limit: int = 10):
        return {"skip": skip, "limit": min(limit, max_limit)}
    return inner

@app.get("/items/")
async def read_items(pagination: Annotated[dict, Depends(pagination_params(50))]):
    return pagination

Callable Class Dependency

class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False

checker = FixedContentQueryChecker(fixed_content="bar")

@app.get("/query-checker/")
async def check_query(fixed_content_included: Annotated[bool, Depends(checker)]):
    return {"fixed_content_in_query": fixed_content_included}

Instance with __call__ allows storing state between requests.

Use Cases

• Database sessions
• Authentication/Authorization
• Rate limiting
• Logging
• Shared query parameters
• Feature flags
• Configuration injection

````

Error Handling

Return HTTP errors to clients with proper status codes.

HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

Key points:

• raise not return - terminates request immediately
• detail can be any JSON-serializable value (str, dict, list)

Response Format

{
  "detail": "Item not found"
}

Custom Headers

raise HTTPException(
    status_code=404,
    detail="Item not found",
    headers={"X-Error": "Custom header value"}
)

Custom Exception Handler

from fastapi import Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong"}
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

Override Validation Errors

from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

Reuse Default Handlers

from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)

@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"HTTP error: {repr(exc)}")  # Log it
    return await http_exception_handler(request, exc)  # Use default

Common Status Codes

FastAPI vs Starlette HTTPException

• FastAPI's HTTPException accepts any JSON-able detail
• Starlette's only accepts strings
• Register handlers for Starlette's version to catch all:

from starlette.exceptions import HTTPException as StarletteHTTPException

First Steps

Core concepts for creating a minimal FastAPI application.

Minimal Application

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

Key Components

1. FastAPI Instance

• app = FastAPI() creates the main application instance
• This is the entry point for all API functionality
• FastAPI inherits from Starlette (all Starlette features available)

2. Path Operation Decorator

• @app.get("/") defines a route handler
• Supports all HTTP methods:
• @app.get() - read data
• @app.post() - create data
• @app.put() - update data
• @app.delete() - delete data
• @app.patch(), @app.options(), @app.head(), @app.trace()

3. Path Operation Function

• Can be async def or regular def
• Returns dict, list, str, int, Pydantic models
• Automatic JSON serialization

Running the Server

fastapi dev main.py

Server runs at http://127.0.0.1:8000

Auto-Generated Documentation

OpenAPI Integration

• FastAPI auto-generates OpenAPI 3.1.0 schema
• Schema includes paths, parameters, request/response models
• Powers interactive documentation
• Can generate client SDKs

Terminology

Forms and File Uploads

Form data handling and file uploads in FastAPI.

Form Data

from typing import Annotated
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    return {"username": username}

Key Points

• Dependency: pip install python-multipart
• Import: from fastapi import Form
• Encoding: application/x-www-form-urlencoded
• Validation: Same as Body(), Query(), Path()

OAuth2 Password Flow

# OAuth2 spec requires exact field names
@app.post("/token")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    # Authenticate user
    return {"access_token": token}

File Uploads

bytes Parameter

from typing import Annotated
from fastapi import FastAPI, File

@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}

• Entire file stored in memory
• Good for small files only

UploadFile Parameter

from fastapi import FastAPI, UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

UploadFile Attributes

Async Methods

contents = await myfile.read()      # Read all bytes
await myfile.seek(0)                # Go to start
await myfile.write(data)            # Write bytes
await myfile.close()                # Close file

Sync Access (in def functions)

contents = myfile.file.read()

Optional File Upload

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No file sent"}
    return {"filename": file.filename}

File with Metadata

@app.post("/uploadfile/")
async def create_upload_file(
    file: Annotated[UploadFile, File(description="A file read as UploadFile")]
):
    return {"filename": file.filename}

Multiple File Uploads

from fastapi import FastAPI, File, UploadFile

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}

# With metadata
@app.post("/uploadfiles/")
async def create_upload_files(
    files: Annotated[list[UploadFile], File(description="Multiple files")]
):
    return {"filenames": [file.filename for file in files]}

Forms and Files Together

from fastapi import FastAPI, File, Form, UploadFile

@app.post("/files/")
async def create_file(
    file: Annotated[bytes, File()],
    fileb: Annotated[UploadFile, File()],
    token: Annotated[str, Form()]
):
    return {
        "file_size": len(file),
        "fileb_content_type": fileb.content_type,
        "token": token
    }

Encoding Types

Prohibitions

• ❌ Cannot mix Body (JSON) with Form/File in same endpoint
• ❌ Don't use bytes for large files (memory)

Recipes

File Upload with Validation

from fastapi import HTTPException

ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
MAX_SIZE = 5 * 1024 * 1024  # 5MB

@app.post("/upload/")
async def upload_file(file: UploadFile):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, "Invalid file type")

    contents = await file.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(400, "File too large")

    # Process file...
    return {"filename": file.filename}

Save Uploaded File

import shutil
from pathlib import Path

UPLOAD_DIR = Path("uploads")

@app.post("/upload/")
async def upload_file(file: UploadFile):
    file_path = UPLOAD_DIR / file.filename
    with file_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename, "path": str(file_path)}

Lifespan Events

Startup/shutdown logic for FastAPI applications.

Lifespan Context Manager (Recommended)

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: runs before accepting requests
    print("Starting up...")
    ml_model = load_model()
    app.state.model = ml_model

    yield  # Application runs here

    # Shutdown: runs when shutting down
    print("Shutting down...")
    ml_model.unload()

app = FastAPI(lifespan=lifespan)

@app.get("/predict")
def predict(x: float):
    return {"result": app.state.model.predict(x)}

Shared Resources Pattern

from contextlib import asynccontextmanager

# Global dict for resources
resources = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Setup
    resources["db"] = await create_db_pool()
    resources["cache"] = await create_redis_connection()

    yield

    # Cleanup
    await resources["db"].close()
    await resources["cache"].close()

app = FastAPI(lifespan=lifespan)

@app.get("/users")
async def get_users():
    db = resources["db"]
    return await db.fetch("SELECT * FROM users")

Database Connection Example

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine("postgresql+asyncpg://...")
    app.state.engine = engine

    yield

    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Testing Lifespan Events

from fastapi.testclient import TestClient

def test_lifespan():
    with TestClient(app) as client:
        # Startup runs when entering context
        response = client.get("/predict")
        assert response.status_code == 200
    # Shutdown runs when exiting context

For async tests:

import pytest
from httpx import ASGITransport, AsyncClient
from asgi_lifespan import LifespanManager

@pytest.mark.anyio
async def test_async():
    async with LifespanManager(app):
        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://test"
        ) as client:
            response = await client.get("/")

Requires: pip install asgi-lifespan

Deprecated: Event Handlers

# Old style - deprecated, use lifespan instead
@app.on_event("startup")
async def startup():
    print("Starting...")

@app.on_event("shutdown")
async def shutdown():
    print("Stopping...")

Notes

• Lifespan context manager is the modern approach
• yield separates startup from shutdown
• Use app.state to store shared resources
• Cleanup in shutdown prevents resource leaks
• TestClient automatically handles lifespan events

API Metadata and Documentation URLs

Customize OpenAPI metadata and documentation UI.

API Metadata

from fastapi import FastAPI

description = """
ChimichangApp API helps you do awesome stuff. 🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""

app = FastAPI(
    title="ChimichangApp",
    description=description,
    summary="Deadpool's favorite app. Nuff said.",
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

Metadata Fields

License Identifier (OpenAPI 3.1.0+)

license_info={
    "name": "Apache 2.0",
    "identifier": "MIT",  # Instead of URL
}

Tag Metadata

tags_metadata = [
    {
        "name": "users",
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]

@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]

Tag order in docs follows list order (not alphabetical).

OpenAPI URL

# Default: /openapi.json
app = FastAPI(openapi_url="/api/v1/openapi.json")

# Disable OpenAPI entirely
app = FastAPI(openapi_url=None)

Docs URLs

# Defaults: /docs (Swagger), /redoc (ReDoc)
app = FastAPI(
    docs_url="/documentation",  # Swagger UI
    redoc_url="/redocumentation",  # ReDoc
)

# Disable docs
app = FastAPI(docs_url=None, redoc_url=None)

Static Files

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")

Mount Parameters

What is Mounting?

Mounting adds a complete independent application at a specific path. Mounted apps are:

• Completely independent
• Not included in main app's OpenAPI
• Handle all sub-paths themselves

Debugging

VS Code / PyCharm Debug Setup

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

VS Code Debug Config

1. Go to "Debug" panel 2. "Add configuration..." → Python 3. Select "Python: Current File (Integrated Terminal)" 4. Set breakpoints and run

PyCharm Debug

1. Open "Run" menu 2. Select "Debug..." 3. Choose the file (e.g., main.py)

About __name__ == "__main__"

• Runs code only when file is executed directly: python myapp.py
• Does NOT run when imported: from myapp import app

Recipes

Production vs Development Docs

import os

ENVIRONMENT = os.getenv("ENVIRONMENT", "development")

app = FastAPI(
    docs_url="/docs" if ENVIRONMENT == "development" else None,
    redoc_url="/redoc" if ENVIRONMENT == "development" else None,
)

Custom OpenAPI Schema

from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title="Custom Title",
        version="1.0.0",
        description="Custom description",
        routes=app.routes,
    )
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Serve Static with HTML

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def read_index():
    return FileResponse("static/index.html")

Middleware

Functions that process every request and response.

How Middleware Works

1. Receive incoming request 2. Execute code before route handler 3. Pass request to route 4. Receive response from route 5. Execute code after route handler 6. Return response

Create Middleware

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Middleware Function Parameters

• request: Request - The incoming request
• call_next - Function that passes request to route and returns response

Execution Order

Middleware is stacked (last added = outermost):

app.add_middleware(MiddlewareA)
app.add_middleware(MiddlewareB)

Execution:

• Request: B → A → route
• Response: route → A → B

Common Patterns

Request Logging

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response

Authentication Check

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    if request.url.path.startswith("/protected"):
        auth = request.headers.get("Authorization")
        if not auth:
            return JSONResponse(status_code=401, content={"detail": "Unauthorized"})
    return await call_next(request)

Exception Handling

@app.middleware("http")
async def catch_exceptions(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        return JSONResponse(status_code=500, content={"detail": str(e)})

Built-in Middleware

CORS

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

GZip Compression

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

Trusted Host

from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

HTTPS Redirect

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(HTTPSRedirectMiddleware)

ASGI Middleware

Use any ASGI-compatible middleware:

from unicorn import UnicornMiddleware

# Simple wrapping (not recommended)
# new_app = UnicornMiddleware(app, some_config="value")

# Recommended: use add_middleware for proper error handling
app.add_middleware(UnicornMiddleware, some_config="rainbow")

Third-Party ASGI Middleware

• ProxyHeadersMiddleware (uvicorn)
• MessagePack middleware
• See: ASGI Awesome List

Timing Notes

• Dependencies with yield: exit code runs after middleware
• Background tasks: run after all middleware completes
• Use time.perf_counter() for precise timing (not time.time())

Custom Headers

• Prefix custom headers with X-
• For browser-visible headers, configure in CORS expose_headers

````markdown

Pydantic Models

Response models, multiple model patterns, and data transformation.

Return Type Annotation

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    tags: list[str] = []

@app.post("/items/")
async def create_item(item: Item) -> Item:
    return item

@app.get("/items/")
async def read_items() -> list[Item]:
    return [Item(name="Foo", price=42.0)]

FastAPI uses return type for:

• Data validation
• JSON Schema in OpenAPI
• Output data filtering

response_model Parameter

Use when return type differs from actual response:

class UserIn(BaseModel):
    username: str
    password: str  # Don't expose!
    email: str

class UserOut(BaseModel):
    username: str
    email: str

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user  # Password filtered out automatically

Multiple Models Pattern

Different model states for different contexts:

from pydantic import BaseModel, EmailStr

# Base model (shared fields)
class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

# Input model
class UserIn(UserBase):
    password: str

# Output model (no password)
class UserOut(UserBase):
    pass

# Database model
class UserInDB(UserBase):
    hashed_password: str

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    hashed = hash_password(user_in.password)
    user_db = UserInDB(**user_in.model_dump(), hashed_password=hashed)
    return user_db

Model Conversion

# Convert to dict
user_dict = user_in.model_dump()

# Create model with extra fields
UserInDB(**user_in.model_dump(), hashed_password=hashed)

# Partial copy with updates
updated = stored_model.model_copy(update=update_data)

0.136.1 note:

• FastAPI refreshed its own Pydantic v2 integration to avoid deprecated patterns. In app code, prefer current Pydantic v2 methods such as model_dump() / model_copy() and avoid reviving deprecated v1-style helpers in new examples.

Response Filtering Options

# Exclude unset values
@app.get("/items/{id}", response_model=Item, response_model_exclude_unset=True)

# Include only specific fields
@app.get("/items/{id}/name", response_model=Item, response_model_include={"name"})

# Exclude specific fields
@app.get("/items/{id}/public", response_model=Item, response_model_exclude={"tax"})

# Exclude defaults/None
response_model_exclude_defaults=True
response_model_exclude_none=True

Union Types (Multiple Response Types)

from typing import Union

class CarItem(BaseModel):
    type: str = "car"
    description: str

class PlaneItem(BaseModel):
    type: str = "plane"
    size: int
    description: str

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

⚠️ Put more specific type first: Union[PlaneItem, CarItem]

List and Dict Responses

# List of models
@app.get("/items/", response_model=list[Item])
async def read_items():
    return items

# Arbitrary dict
@app.get("/weights/", response_model=dict[str, float])
async def read_weights():
    return {"foo": 2.3, "bar": 3.4}

Return Response Directly

from fastapi.responses import JSONResponse, RedirectResponse

@app.get("/portal")
async def get_portal(teleport: bool = False):
    if teleport:
        return RedirectResponse(url="https://example.com")
    return JSONResponse(content={"message": "Here"})

# Disable response model validation
@app.get("/portal", response_model=None)
async def get_portal():
    return {"message": "No validation"}

CRUD Model Pattern

class ItemBase(BaseModel):
    name: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None

class ItemInDB(ItemBase):
    id: int
    owner_id: int

class ItemResponse(ItemInDB):
    pass

Partial Updates (PATCH)

@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    stored = items[item_id]
    update_data = item.model_dump(exclude_unset=True)  # Only sent fields
    updated = stored.model_copy(update=update_data)
    items[item_id] = updated
    return updated

Key: exclude_unset=True includes only fields explicitly set by client.

Prohibitions

• ❌ Don't store plaintext passwords in models
• ❌ Don't return input models with sensitive data
• ❌ Don't use PlaneItem | CarItem in response_model (use Union[])

````

OpenAPI Advanced

Callbacks, webhooks, custom operation IDs, SDK generation.

OpenAPI Callbacks

Document external API callbacks your API will trigger:

from fastapi import FastAPI, APIRouter
from pydantic import BaseModel, HttpUrl

app = FastAPI()

class Invoice(BaseModel):
    id: str
    customer: str
    total: float

class InvoiceEvent(BaseModel):
    description: str
    paid: bool

class InvoiceEventReceived(BaseModel):
    ok: bool

# Create callback router (documentation only)
callback_router = APIRouter()

@callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}",
    response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    """External API should implement this endpoint"""
    pass  # No code needed - documentation only

# Main endpoint with callbacks
@app.post("/invoices/", callbacks=callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None):
    """
    Create invoice. Will call callback_url when payment processed.
    """
    return {"msg": "Invoice created"}

Callback path expression: {$callback_url} from query param, {$request.body.id} from request body.

OpenAPI Webhooks

Document webhooks your API sends (OpenAPI 3.1):

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Subscription(BaseModel):
    username: str
    plan: str

@app.webhooks.post("new-subscription")
def new_subscription(body: Subscription):
    """
    Webhook sent when new user subscribes.
    External systems should implement this endpoint.
    """
    pass

Shows in /docs under Webhooks section.

Custom Operation IDs

Clean method names for SDK generation:

from fastapi import FastAPI
from fastapi.routing import APIRoute

def custom_generate_unique_id(route: APIRoute) -> str:
    return f"{route.tags[0]}-{route.name}"

app = FastAPI(generate_unique_id_function=custom_generate_unique_id)

@app.get("/items/", tags=["items"])
def get_items():  # Operation ID: items-get_items
    return []

Path Operation Config

@app.get(
    "/items/{item_id}",
    operation_id="get_single_item",  # Custom ID
    summary="Get Item",
    description="Retrieve item by ID",
    deprecated=True,  # Mark as deprecated
    include_in_schema=True  # Set False to hide
)
def read_item(item_id: str):
    return {"id": item_id}

Exclude from OpenAPI

@app.get("/internal", include_in_schema=False)
def internal_endpoint():
    return {"secret": "data"}

OpenAPI Extra

Add arbitrary OpenAPI fields:

@app.post(
    "/items/",
    openapi_extra={
        "x-custom-field": "value",
        "requestBody": {
            "content": {
                "application/x-yaml": {
                    "schema": {"type": "object"}
                }
            },
            "required": True
        }
    }
)
def create_item():
    return {}

SDK Generation

Using Hey API (TypeScript)

npx @hey-api/openapi-ts -i http://localhost:8000/openapi.json -o src/client

Using OpenAPI Generator

openapi-generator generate \
  -i http://localhost:8000/openapi.json \
  -g python \
  -o ./client

Preprocess OpenAPI for Cleaner Names

import json
from pathlib import Path

# Download openapi.json first
file = Path("openapi.json")
spec = json.loads(file.read_text())

# Remove tag prefix from operation IDs
for path_data in spec["paths"].values():
    for operation in path_data.values():
        tag = operation["tags"][0]
        op_id = operation["operationId"]
        # items-get_items -> get_items
        operation["operationId"] = op_id.removeprefix(f"{tag}-")

file.write_text(json.dumps(spec))

Sub-Applications (Mounts)

Independent apps with separate OpenAPI:

app = FastAPI()
subapi = FastAPI()

@app.get("/app")
def main_root():
    return {"message": "Main app"}

@subapi.get("/sub")
def sub_root():
    return {"message": "Sub app"}

app.mount("/subapi", subapi)
# Main docs: /docs
# Sub docs: /subapi/docs

WSGI Integration (Flask/Django)

from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware
from flask import Flask

flask_app = Flask(__name__)

@flask_app.route("/")
def flask_root():
    return "Hello from Flask"

app = FastAPI()

@app.get("/v2")
def fastapi_root():
    return {"message": "FastAPI"}

app.mount("/v1", WSGIMiddleware(flask_app))
# /v1/* -> Flask
# /v2   -> FastAPI

Path Operation Configuration

Metadata and configuration for endpoint documentation.

Response Status Code

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

# Or use integer directly
@app.post("/items/", status_code=201)
async def create_item(item: Item):
    return item

Common codes: 200 OK, 201 Created, 204 No Content, 404 Not Found.

See error-handling.md for full status codes table.

Tags

Group endpoints in docs:

@app.post("/items/", tags=["items"])
async def create_item(item: Item):
    return item

@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]

Tags with Enum

from enum import Enum

class Tags(Enum):
    items = "items"
    users = "users"

@app.get("/items/", tags=[Tags.items])
async def get_items():
    return ["Portal gun", "Plumbus"]

Summary and Description

@app.post(
    "/items/",
    summary="Create an item",
    description="Create an item with all the information: name, description, price, tax, tags"
)
async def create_item(item: Item):
    return item

Description from Docstring

@app.post("/items/", summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

Markdown supported in docstrings!

Response Description

@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    response_description="The created item"
)
async def create_item(item: Item):
    """Create an item with all the information."""
    return item

OpenAPI requires response description. FastAPI defaults to "Successful response" if omitted.

Deprecate Endpoint

@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]

Shown with strikethrough in docs.

JSON Compatible Encoder

Convert Pydantic models to JSON-compatible dicts:

from datetime import datetime
from fastapi.encoders import jsonable_encoder

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_data = jsonable_encoder(item)
    # datetime converted to ISO string
    fake_db[id] = json_compatible_data

What jsonable_encoder Does

• datetime → ISO format string
• Pydantic model → dict
• Recursively converts nested objects
• Returns Python dict (not JSON string)

Body Updates

Full Update (PUT)

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    stored = items[item_id]
    stored_model = Item(**stored)
    update_data = item.model_dump()
    updated = stored_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated)
    return updated

Partial Update (PATCH)

@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    stored = items[item_id]
    stored_model = Item(**stored)
    update_data = item.model_dump(exclude_unset=True)  # Only sent fields
    updated = stored_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated)
    return updated

Key: exclude_unset=True includes only fields explicitly set by client.

All Configuration Parameters

@app.post(
    "/items/",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    tags=["items"],
    summary="Create an item",
    description="Create a new item",
    response_description="The created item",
    deprecated=False,
    operation_id="create_item_items_post",
    include_in_schema=True,
    responses={
        201: {"description": "Created successfully"},
        400: {"description": "Bad request"}
    }
)
async def create_item(item: Item):
    return item

OpenAPI Operation ID

# Auto-generated: create_item_items_post
@app.post("/items/", operation_id="createItem")
async def create_item(item: Item):
    return item

Useful for SDK generation.

Exclude from Schema

@app.get("/internal/", include_in_schema=False)
async def internal_endpoint():
    return {"internal": "data"}

Not shown in OpenAPI docs.

Path Parameters

Capture dynamic values from URL paths.

Basic Syntax

@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

Type Annotations

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Benefits:

• Data conversion: String "3" → integer 3
• Data validation: Invalid values return clear error
• Editor support: Autocomplete, type checking
• Auto documentation: Types shown in OpenAPI docs

Validation Error Response

{
  "detail": [
    {
      "type": "int_parsing",
      "loc": ["path", "item_id"],
      "msg": "Input should be a valid integer",
      "input": "foo"
    }
  ]
}

Order Matters

Fixed paths must be declared before parameterized paths:

# ✅ Correct order
@app.get("/users/me")  # First - specific
async def read_user_me():
    return {"user_id": "current user"}

@app.get("/users/{user_id}")  # Second - generic
async def read_user(user_id: str):
    return {"user_id": user_id}

Predefined Values with Enum

from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning"}
    return {"model_name": model_name}

Working with Enum values:

• Compare: model_name is ModelName.alexnet
• Get value: model_name.value → "alexnet"
• Returns JSON serialized string

Path Parameters Containing Paths

Use :path converter for file paths:

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

URL: /files//home/user/file.txt (note double slash)

Supported Types

• str, int, float, bool
• UUID
• Custom types via Pydantic

Query Parameters

Parameters passed via URL query string (?key=value&key2=value2).

Basic Usage

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]

URL: http://127.0.0.1:8000/items/?skip=20&limit=5

Parameter Types

Required Parameters

@app.get("/items/{item_id}")
async def read_item(item_id: str, needy: str):
    return {"item_id": item_id, "needy": needy}

No default = required. Missing parameter returns error.

Optional Parameters

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

Default Values

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]

Boolean Conversion

FastAPI auto-converts truthy strings to bool:

@app.get("/items/{item_id}")
async def read_item(item_id: str, short: bool = False):
    ...

All these evaluate to True:

• ?short=1
• ?short=True
• ?short=true
• ?short=on
• ?short=yes

Mixed Parameters

Combine path and query parameters freely:

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int,           # path
    item_id: str,           # path
    q: str | None = None,   # query, optional
    short: bool = False     # query, default
):
    ...

FastAPI distinguishes by:

• Path parameters: Declared in route path {param}
• Query parameters: Other function parameters

Common Patterns

Pagination

@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 100):
    return db.query(Item).offset(skip).limit(limit).all()

Filtering

@app.get("/items/")
async def search_items(
    q: str | None = None,
    category: str | None = None,
    min_price: float | None = None
):
    ...

Required + Optional Mix

@app.get("/search/")
async def search(
    query: str,                    # required
    page: int = 1,                 # default
    per_page: int | None = None    # optional
):
    ...

Request Body

Send JSON data from client to API using Pydantic models.

Basic Example

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item

Pydantic Model Rules

• Required fields: no default value
• Optional fields: = None
• Fields with defaults: = value

Valid JSON for above model:

{"name": "Foo", "price": 45.2}
{"name": "Foo", "description": "A thing", "price": 45.2, "tax": 3.5}

What FastAPI Does Automatically

1. Reads request body as JSON 2. Converts types (string → int, etc.) 3. Validates data structure 4. Returns clear errors for invalid data 5. Provides editor autocomplete for model attributes 6. Generates JSON Schema for OpenAPI docs

Using the Model

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.model_dump()  # Convert to dict
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

Combining Parameters

FastAPI distinguishes parameter types automatically:

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,              # Path parameter
    item: Item,                # Request body (Pydantic model)
    q: str | None = None       # Query parameter
):
    return {"item_id": item_id, **item.model_dump(), "q": q}

Recognition rules:

• Declared in path → path parameter
• Pydantic model → request body
• Singular type (str, int, etc.) → query parameter

HTTP Methods for Body

• POST - Create (most common)
• PUT - Update/Replace
• PATCH - Partial update
• DELETE - Delete (rarely has body)

⚠️ GET with body is discouraged (undefined behavior in specs).

Advanced Responses

Custom responses, status codes, cookies, headers for FastAPI.

Return Response Directly

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

@app.get("/item/")
def read_item():
    # Use jsonable_encoder for datetime, Pydantic models etc
    data = jsonable_encoder({"created": datetime.now()})
    return JSONResponse(content=data)

Custom Response Types

````python from fastapi.responses import ( HTMLResponse, # text/html PlainTextResponse, # text/plain JSONResponse, # application/json (default) ORJSONResponse, # Fast JSON (pip install orjson) UJSONResponse, # Fast JSON (pip install ujson) RedirectResponse, # HTTP redirect StreamingResponse, # Streaming content FileResponse # File downloads )

HTML Response

@app.get("/", response_class=HTMLResponse) def get_html(): return "<h1>Hello</h1>"

Streaming Response

@app.get("/stream") def stream(): def generate(): for i in range(10): yield f"chunk {i}\n" return StreamingResponse(generate(), media_type="text/plain")

File Response

@app.get("/file") def get_file(): return FileResponse( path="report.pdf", filename="download.pdf", media_type="application/pdf" )

Redirect

@app.get("/old") def redirect(): return RedirectResponse(url="/new", status_code=301)

Streaming with yield (0.134.0+)

FastAPI can stream incrementally without buffering the whole response. Choose the pattern based on the client protocol.

Stream JSON Lines (JSONL)

Use yield to emit one JSON item per line (content type is application/jsonl). If you annotate the return type (e.g. AsyncIterable[Item]), FastAPI will validate/serialize via Pydantic and document the schema.

from collections.abc import AsyncIterable
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None

@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for i in range(3):
        yield Item(name=f"item-{i}")

Notes:

• JSON Lines is not the same as a JSON array; clients must parse line-by-line.
• If you omit the return type annotation, FastAPI falls back to jsonable_encoder for serialization.

Stream bytes / binary data

For raw bytes (audio/video/chunks), use StreamingResponse and set media_type. If the streaming code is blocking (disk/network reads), prefer def so FastAPI can run it in a threadpool.

from collections.abc import Iterable
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/audio")
def stream_audio() -> StreamingResponse:
    def iter_bytes() -> Iterable[bytes]:
        yield b"..."  # chunk 1
        yield b"..."  # chunk 2
    return StreamingResponse(iter_bytes(), media_type="audio/mpeg")

Server-Sent Events (SSE) (0.135.0+)

SSE uses text/event-stream and is browser-native via EventSource. In FastAPI, set response_class=EventSourceResponse and yield items.

from collections.abc import AsyncIterable
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()

@app.get("/events", response_class=EventSourceResponse)
async def events() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream start")
    for i in range(3):
        yield ServerSentEvent(data={"i": i}, event="tick", id=str(i), retry=5000)

Operational notes:

• ServerSentEvent.data is JSON-encoded; use raw_data for raw text payloads.
• Clients can resume after disconnect using Last-Event-ID.
• For proxy behavior: consider keep-alive pings, Cache-Control: no-cache, and disabling buffering (e.g. X-Accel-Buffering: no for Nginx).
• 0.136.2+ validates SSE fields more strictly. Prefer building events with ServerSentEvent(...) instead of hand-assembling ad hoc payload dicts, and test any optional event, id, or retry fields you generate dynamically.

````

Default Response Class

from fastapi.responses import ORJSONResponse

# App-wide default
app = FastAPI(default_response_class=ORJSONResponse)

# Endpoint-specific
@app.get("/items/", response_class=ORJSONResponse)
def read_items():
    return [{"id": 1}]

Additional Status Codes

from fastapi import status
from fastapi.responses import JSONResponse

@app.put("/items/{item_id}")
def upsert_item(item_id: str, item: Item):
    if item_id in items:
        items[item_id] = item
        return item  # 200 default
    else:
        items[item_id] = item
        return JSONResponse(
            status_code=status.HTTP_201_CREATED,
            content=jsonable_encoder(item)
        )

Dynamic Status Code

from fastapi import Response

@app.put("/items/{item_id}")
def update_item(item_id: str, response: Response):
    if item_id not in items:
        response.status_code = status.HTTP_201_CREATED
    return {"item_id": item_id}

Response Cookies

from fastapi import Response

# Via Response parameter
@app.post("/login")
def login(response: Response):
    response.set_cookie(
        key="session",
        value="abc123",
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=3600
    )
    return {"message": "logged in"}

# Via direct Response
@app.post("/login")
def login():
    response = JSONResponse(content={"message": "logged in"})
    response.set_cookie(key="session", value="abc123")
    return response

# Delete cookie
@app.post("/logout")
def logout(response: Response):
    response.delete_cookie(key="session")
    return {"message": "logged out"}

Response Headers

from fastapi import Response

# Via Response parameter
@app.get("/items/")
def read_items(response: Response):
    response.headers["X-Custom"] = "value"
    response.headers["Cache-Control"] = "max-age=3600"
    return {"items": []}

# Via direct Response
@app.get("/items/")
def read_items():
    return JSONResponse(
        content={"items": []},
        headers={"X-Custom": "value"}
    )

Note: Custom headers with X- prefix. For CORS, expose custom headers in CORSMiddleware(expose_headers=[...]).

Multiple Response Docs (OpenAPI)

from fastapi import HTTPException
from pydantic import BaseModel

class Item(BaseModel):
    name: str

class Message(BaseModel):
    message: str

@app.get(
    "/items/{item_id}",
    response_model=Item,
    responses={
        200: {"model": Item, "description": "Item found"},
        404: {"model": Message, "description": "Item not found"},
        422: {"description": "Validation error"}
    }
)
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

Combine Response Info

responses = {
    404: {"description": "Not found"},
    403: {"description": "Forbidden"}
}

@app.get("/users/{user_id}", responses={**responses, 200: {"model": User}})
def read_user(user_id: int):
    ...

Security

Authentication and authorization patterns in FastAPI.

Security Schemes (OpenAPI)

FastAPI supports standard OpenAPI security schemes:

OAuth2 Password Flow (Simple)

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    # Validate user credentials
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=400, detail="Incorrect credentials")
    return {"access_token": user.username, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(token: Annotated[str, Depends(oauth2_scheme)]):
    user = get_user_from_token(token)
    return user

Get Current User Pattern

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    user = decode_token(token)
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_user)]
):
    return current_user

JWT Tokens

from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

Password Hashing

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

HTTP Basic Auth

from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

security = HTTPBasic()

def get_current_username(
    credentials: Annotated[HTTPBasicCredentials, Depends(security)]
):
    # Use compare_digest to prevent timing attacks
    correct_user = secrets.compare_digest(
        credentials.username.encode("utf8"),
        b"admin"
    )
    correct_pass = secrets.compare_digest(
        credentials.password.encode("utf8"),
        b"secret"
    )
    if not (correct_user and correct_pass):
        raise HTTPException(
            status_code=401,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Basic"}
        )
    return credentials.username

@app.get("/admin")
def admin_area(username: Annotated[str, Depends(get_current_username)]):
    return {"message": f"Hello {username}"}

Important: Always use secrets.compare_digest() - regular == is vulnerable to timing attacks.

OAuth2 Scopes

Fine-grained permissions with scopes:

from fastapi import Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "items:read": "Read items",
        "items:write": "Create/update items",
        "users:read": "Read user info"
    }
)

async def get_current_user(
    security_scopes: SecurityScopes,
    token: Annotated[str, Depends(oauth2_scheme)]
):
    # Build WWW-Authenticate header
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"

    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value}
    )

    # Decode and verify token
    payload = verify_token(token)
    if payload is None:
        raise credentials_exception

    # Check scopes
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(
                status_code=403,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value}
            )

    return get_user(payload["sub"])

# Use Security() instead of Depends() for scopes
@app.get("/items/")
async def read_items(
    user: Annotated[User, Security(get_current_user, scopes=["items:read"])]
):
    return items

@app.post("/items/")
async def create_item(
    item: Item,
    user: Annotated[User, Security(get_current_user, scopes=["items:write"])]
):
    return item

Token with Scopes

def create_access_token(data: dict, scopes: list[str]):
    to_encode = data.copy()
    to_encode.update({
        "scopes": scopes,
        "exp": datetime.utcnow() + timedelta(minutes=30)
    })
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400)

    # Grant requested scopes (validate against user permissions)
    token = create_access_token(
        data={"sub": user.username},
        scopes=form_data.scopes
    )
    return {"access_token": token, "token_type": "bearer"}

Settings & Environment Variables

Configuration management with Pydantic Settings.

Install

pip install pydantic-settings

Basic Settings Class

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    admin_email: str
    items_per_user: int = 50

settings = Settings()  # Reads from env vars

# Usage
@app.get("/info")
def info():
    return {"app_name": settings.app_name}

Environment Variables

Variables are case-insensitive:

# Set env vars
export ADMIN_EMAIL="admin@example.com"
export APP_NAME="MyApp"

# Or inline
ADMIN_EMAIL="admin@example.com" fastapi run main.py

Settings in Separate Module

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool = False

settings = Settings()

# main.py
from .config import settings

app = FastAPI(debug=settings.debug)

Settings as Dependency

from functools import lru_cache
from fastapi import Depends

class Settings(BaseSettings):
    api_key: str
    database_url: str

@lru_cache  # Create once, reuse
def get_settings():
    return Settings()

@app.get("/config")
def read_config(settings: Annotated[Settings, Depends(get_settings)]):
    return {"key": settings.api_key}

.env File Support

# .env
DATABASE_URL="postgresql://localhost/db"
SECRET_KEY="mysecret"
DEBUG=true

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool = False

    model_config = SettingsConfigDict(env_file=".env")

Requires: pip install python-dotenv

Nested Settings

class DatabaseSettings(BaseSettings):
    host: str = "localhost"
    port: int = 5432
    name: str = "app"

class Settings(BaseSettings):
    db: DatabaseSettings = DatabaseSettings()

    model_config = SettingsConfigDict(env_nested_delimiter="__")

# Set via: DB__HOST=myhost DB__PORT=5433

Override in Tests

# test_main.py
from fastapi.testclient import TestClient
from .config import Settings
from .main import app, get_settings

def get_settings_override():
    return Settings(admin_email="test@example.com")

app.dependency_overrides[get_settings] = get_settings_override

client = TestClient(app)

def test_app():
    response = client.get("/info")
    assert response.json()["admin_email"] == "test@example.com"

lru_cache Explained

@lru_cache
def get_settings():
    return Settings()  # Only called once

Without @lru_cache, each request reads .env file again. With it, settings loaded once and cached.

Validation

Pydantic validates types automatically:

class Settings(BaseSettings):
    port: int  # Must be integer
    allowed_hosts: list[str] = ["localhost"]

# ALLOWED_HOSTS='["host1.com", "host2.com"]'

Sensitive Values

from pydantic import SecretStr

class Settings(BaseSettings):
    password: SecretStr

settings = Settings()
print(settings.password)  # SecretStr('**********')
print(settings.password.get_secret_value())  # actual value

SQL Databases

Use SQLModel with FastAPI for SQL database integration.

Install

pip install sqlmodel

SQLModel = SQLAlchemy + Pydantic (same author as FastAPI).

Basic Model

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str

Key points:

• table=True = database table model
• Field(primary_key=True) = primary key
• Field(index=True) = create SQL index

Engine Setup

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}  # Required for SQLite
engine = create_engine(sqlite_url, connect_args=connect_args)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

Session Dependency

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

Create Tables on Startup

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

CRUD Operations

Create

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

Read List

@app.get("/heroes/")
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

Read One

@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

Delete

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

Multiple Models Pattern

Separate concerns with inheritance:

# Base model (shared fields)
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

# Table model (database)
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

# Public model (API response - no secret_name)
class HeroPublic(HeroBase):
    id: int

# Create model (API input)
class HeroCreate(HeroBase):
    secret_name: str

# Update model (partial updates)
class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None

Using Multiple Models

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db

Supported Databases

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server
• Any SQLAlchemy-supported database

Production Notes

• Use Alembic for migrations
• PostgreSQL recommended for production
• Check Full Stack FastAPI Template

Templates

Jinja2 HTML templates in FastAPI.

Install

pip install jinja2

Basic Setup

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/", response_class=HTMLResponse)
def read_root(request: Request):
    return templates.TemplateResponse(
        request=request,
        name="index.html",
        context={"title": "Home", "items": ["a", "b", "c"]}
    )

Project Structure

project/
├── main.py
├── templates/
│   ├── base.html
│   └── index.html
└── static/
    └── styles.css

Template File (templates/index.html)

<!DOCTYPE html>
<html>
  <head>
    <title>{{ title }}</title>
    <link href="{{ url_for('static', path='styles.css') }}" rel="stylesheet" />
  </head>
  <body>
    <h1>{{ title }}</h1>
    <ul>
      {% for item in items %}
      <li>{{ item }}</li>
      {% endfor %}
    </ul>
  </body>
</html>

url_for in Templates

<!-- Static files -->
<link href="{{ url_for('static', path='css/style.css') }}" rel="stylesheet" />
<img src="{{ url_for('static', path='images/logo.png') }}" />

<!-- Route links -->
<a href="{{ url_for('read_item', item_id=1) }}">Item 1</a>

Static Files Setup

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

Template Inheritance

<!-- templates/base.html -->
<!DOCTYPE html>
<html>
  <head>
    <title>{% block title %}{% endblock %} - My Site</title>
  </head>
  <body>
    <nav>...</nav>
    {% block content %}{% endblock %}
  </body>
</html>

<!-- templates/page.html -->
{% extends "base.html" %} {% block title %}Page Title{% endblock %} {% block content %}
<h1>Page Content</h1>
{% endblock %}

Passing Data

@app.get("/users/{user_id}", response_class=HTMLResponse)
def user_page(request: Request, user_id: int):
    user = get_user(user_id)
    return templates.TemplateResponse(
        request=request,
        name="user.html",
        context={
            "user": user,
            "is_admin": user.role == "admin"
        }
    )

Request Object Required

Template requires request in context for url_for() to work:

# Correct
return templates.TemplateResponse(
    request=request,  # Required
    name="page.html",
    context={"data": data}
)

Custom Response Headers

@app.get("/", response_class=HTMLResponse)
def home(request: Request):
    return templates.TemplateResponse(
        request=request,
        name="home.html",
        context={},
        headers={"X-Custom": "value"}
    )

Jinja2 Features

<!-- Conditionals -->
{% if user.is_admin %}
<span>Admin</span>
{% endif %}

<!-- Loops -->
{% for item in items %}
<li>{{ item.name }}</li>
{% endfor %}

<!-- Filters -->
{{ name|title }} {{ date|strftime('%Y-%m-%d') }}

<!-- Include -->
{% include "partials/header.html" %}

Testing

Test FastAPI applications using TestClient and pytest.

Setup

pip install httpx pytest

Basic Test

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

Key Points

• Test functions use def, not async def
• Client calls are synchronous (no await)
• TestClient is from Starlette, re-exported by FastAPI

0.136.1 note:

• FastAPI now rides on Starlette 1.0.0. If tests rely on Starlette internals or older TestClient behavior beyond the FastAPI re-exported API, re-test those assumptions after upgrading.

Project Structure

app/
├── __init__.py
├── main.py
└── test_main.py

# test_main.py
from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200

Request Methods

# GET with query params
response = client.get("/items/?skip=0&limit=10")

# POST with JSON body
response = client.post(
    "/items/",
    json={"name": "Foo", "price": 42.0}
)

# With headers
response = client.get(
    "/items/foo",
    headers={"X-Token": "secret-token"}
)

# With cookies
response = client.get(
    "/items/",
    cookies={"session": "abc123"}
)

# Form data
response = client.post(
    "/login/",
    data={"username": "user", "password": "pass"}
)

# File upload
with open("file.txt", "rb") as f:
    response = client.post(
        "/upload/",
        files={"file": ("filename.txt", f, "text/plain")}
    )

Testing Errors

def test_read_item_not_found():
    response = client.get("/items/nonexistent")
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}

def test_create_item_invalid():
    response = client.post(
        "/items/",
        json={"name": "Foo"}  # Missing required field
    )
    assert response.status_code == 422  # Validation error

Override Dependencies

from fastapi.testclient import TestClient
from app.main import app, get_db

def override_get_db():
    return TestDatabase()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_with_mock_db():
    response = client.get("/items/")
    assert response.status_code == 200

# Reset overrides after test
app.dependency_overrides = {}

Override Settings

from .config import Settings
from .main import app, get_settings

def get_settings_override():
    return Settings(admin_email="test@example.com")

app.dependency_overrides[get_settings] = get_settings_override

Async Tests

For async database calls or when testing async code:

import pytest
from httpx import ASGITransport, AsyncClient
from .main import app

@pytest.mark.anyio
async def test_async():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200

Requires: pip install anyio pytest-anyio httpx

Note: AsyncClient doesn't trigger lifespan events. Use asgi-lifespan if needed:

from asgi_lifespan import LifespanManager

@pytest.mark.anyio
async def test_with_lifespan():
    async with LifespanManager(app):
        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://test"
        ) as ac:
            response = await ac.get("/")

Testing WebSockets

def test_websocket():
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("hello")
        data = websocket.receive_text()
        assert data == "Echo: hello"

Testing Lifespan Events

def test_app_lifespan():
    with TestClient(app) as client:
        # Startup runs when entering context
        response = client.get("/")
        assert response.status_code == 200
    # Shutdown runs when exiting context

Run Tests

pytest
pytest -v  # Verbose
pytest test_main.py::test_read_main  # Specific test
pytest -x  # Stop on first failure
pytest --tb=short  # Shorter traceback

Parameter Validation

Advanced validation for Query, Path, Header, Cookie parameters.

Query Validation with Query

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(max_length=50)] = None
):
    return {"q": q}

String Validations

q: Annotated[str | None, Query(
    min_length=3,
    max_length=50,
    pattern="^[a-z]+$"  # Regex pattern
)] = None

Numeric Validations (Path/Query)

from fastapi import Path

@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(ge=1, le=1000)],  # 1 ≤ value ≤ 1000
    size: Annotated[float, Query(gt=0, lt=100)]   # 0 < value < 100
):
    ...

Metadata for Documentation

q: Annotated[str | None, Query(
    title="Search query",
    description="Full-text search in item names",
    min_length=3,
    max_length=50,
    deprecated=True  # Shows as deprecated in docs
)] = None

Alias Parameter Name

# URL: /items/?item-query=foo
@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(alias="item-query")] = None
):
    return {"q": q}

Multiple Values (List)

# URL: /items/?q=foo&q=bar
@app.get("/items/")
async def read_items(
    q: Annotated[list[str] | None, Query()] = None
):
    return {"q": q}  # ["foo", "bar"]

With defaults:

q: Annotated[list[str], Query()] = ["default1", "default2"]

Required Parameters with Validation

# Required (no default)
q: Annotated[str, Query(min_length=3)]

# Required but can be None
q: Annotated[str | None, Query(min_length=3)]  # No default!

Custom Validation

from pydantic import AfterValidator

def validate_id(value: str) -> str:
    if not value.startswith(("isbn-", "imdb-")):
        raise ValueError("ID must start with isbn- or imdb-")
    return value

@app.get("/items/")
async def read_items(
    id: Annotated[str, AfterValidator(validate_id)]
):
    ...

Exclude from OpenAPI

hidden: Annotated[str | None, Query(include_in_schema=False)] = None

Best Practice: Use Annotated

# ✅ Recommended (Python 3.9+)
q: Annotated[str | None, Query(max_length=50)] = None

# ❌ Old style (avoid)
q: str | None = Query(default=None, max_length=50)

WebSockets

Real-time bidirectional communication in FastAPI.

Basic WebSocket

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

Client Example (JavaScript)

const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => console.log(event.data);
ws.onopen = () => ws.send("Hello");

Connection Manager (Multiple Clients)

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left")

WebSocket with Dependencies

from fastapi import Query, Depends, WebSocket

async def verify_token(token: str = Query(...)):
    if token != "valid":
        raise WebSocketException(code=1008)
    return token

@app.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Depends(verify_token)
):
    await websocket.accept()
    await websocket.send_text(f"Authenticated with token: {token}")

Handling Disconnection

from fastapi import WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Send/Receive Methods

# Receive
data = await websocket.receive_text()    # str
data = await websocket.receive_bytes()   # bytes
data = await websocket.receive_json()    # dict/list

# Send
await websocket.send_text("hello")       # str
await websocket.send_bytes(b"...")       # bytes
await websocket.send_json({"key": "val"})  # dict/list

# Close
await websocket.close(code=1000)

Testing WebSockets

from fastapi.testclient import TestClient

def test_websocket():
    client = TestClient(app)
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("Hello")
        data = websocket.receive_text()
        assert data == "Echo: Hello"

WebSocket Path Parameters

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    await websocket.accept()
    await websocket.send_text(f"Joined room: {room_id}")

Cookie/Header Authentication

from fastapi import Cookie, Header

@app.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    session: str | None = Cookie(default=None),
    x_token: str | None = Header(default=None)
):
    if not session or not x_token:
        await websocket.close(code=1008)
        return
    await websocket.accept()
    ...