Python FastAPI

A modern, high-performance web framework for building APIs with Python, featuring automatic validation, serialisation, and interactive documentation.

Overview

FastAPI is built on Starlette for web handling and Pydantic for data validation. It leverages Python type hints to provide automatic request validation, response serialisation, and OpenAPI documentation generation. FastAPI is designed for building production-ready APIs quickly whilst maintaining high performance comparable to NodeJS and Go.

⤢ zoom

Creating Endpoints

FastAPI uses decorators to define HTTP endpoints (path operations) with automatic request/response handling.

Key Concepts

• Path operations: Functions decorated with HTTP method decorators (@app.get(), @app.post(), etc.)
• Path operation decorators: Define the URL path and HTTP method
• Response status codes: Set default status codes for successful responses
• Tags: Group endpoints in documentation

Common Patterns

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

# GET endpoint
@app.get("/items")
async def get_items():
    return {"items": []}

# GET with path parameter
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

# POST with status code
@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: dict):
    return item

# PUT endpoint
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: dict):
    return {"item_id": item_id, **item}

# DELETE endpoint
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None

# PATCH endpoint
@app.patch("/items/{item_id}")
async def partial_update_item(item_id: int, item: dict):
    return {"item_id": item_id, **item}

Examples

Organising with APIRouter:

from fastapi import APIRouter, FastAPI

# Create router with prefix and tags
router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def list_users():
    return []

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# Include router in main app
app = FastAPI()
app.include_router(router)

Startup and shutdown with lifespan (replaces the deprecated @app.on_event):

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: runs before the app starts serving requests
    pool = await create_db_pool()
    app.state.db = pool
    yield
    # Shutdown: runs when the app is stopping
    await pool.close()

app = FastAPI(lifespan=lifespan)

Multiple tags and deprecated endpoints:

@app.get(
    "/legacy/items",
    tags=["items", "legacy"],
    deprecated=True,
    summary="Get items (deprecated)",
    description="Use /items instead"
)
async def get_legacy_items():
    return []

Path and Query Parameters

FastAPI automatically extracts and validates parameters from URLs using Python type hints.

Key Concepts

• Path parameters: Variables embedded in the URL path (e.g., /items/{item_id})
• Query parameters: Key-value pairs after ? in URL (e.g., /items?skip=0&limit=10)
• Type conversion: Automatic conversion based on type hints
• Validation: Built-in validation for numbers, strings, and custom types

Common Patterns

from fastapi import FastAPI, Query, Path
from typing import Optional
from enum import Enum

app = FastAPI()

# Path parameters with validation
@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(
        ...,  # Required
        title="Item ID",
        description="The ID of the item to retrieve",
        ge=1,  # Greater than or equal to 1
        le=1000  # Less than or equal to 1000
    )
):
    return {"item_id": item_id}

# Query parameters with defaults
@app.get("/items")
async def list_items(
    skip: int = 0,
    limit: int = Query(default=10, le=100),
    search: Optional[str] = None
):
    return {"skip": skip, "limit": limit, "search": search}

# Required query parameter
@app.get("/search")
async def search_items(
    q: str = Query(..., min_length=3, max_length=50)
):
    return {"query": q}

# Enum for predefined values
class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

@app.get("/sorted-items")
async def get_sorted_items(order: SortOrder = SortOrder.asc):
    return {"order": order}

Examples

Multiple path parameters:

@app.get("/users/{user_id}/items/{item_id}")
async def get_user_item(user_id: int, item_id: int):
    return {"user_id": user_id, "item_id": item_id}

List query parameters:

from typing import List

@app.get("/items/filter")
async def filter_items(
    tags: List[str] = Query(default=[], alias="tag")
):
    # Called as: /items/filter?tag=food&tag=drink
    return {"tags": tags}

Query parameter with regex validation:

@app.get("/users/{username}")
async def get_user(
    username: str = Path(
        ...,
        pattern="^[a-z0-9_-]{3,16}$"
    )
):
    return {"username": username}

Request and Response Models

Pydantic models provide automatic validation, serialisation, and documentation for request/response data.

Key Concepts

• Request body: Data sent in POST/PUT/PATCH requests, validated against Pydantic models
• Response model: Defines the shape of response data, filters out extra fields
• Field validation: Constraints and metadata for model fields
• Model inheritance: Create variations for different operations (create, update, response)

Common Patterns

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field, EmailStr
from typing import Optional
from datetime import datetime, timezone

app = FastAPI()

# Base model with shared fields
class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0)

    model_config = ConfigDict(
        json_schema_extra={
            "example": {
                "name": "Widget",
                "description": "A useful widget",
                "price": 9.99
            }
        }
    )

# Create model (request body)
class ItemCreate(ItemBase):
    pass

# Update model (all fields optional)
class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = None
    price: Optional[float] = Field(None, gt=0)

# Response model (includes computed fields)
class ItemResponse(ItemBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)  # Enable ORM mode

# Using models in endpoints
@app.post("/items", response_model=ItemResponse)
async def create_item(item: ItemCreate):
    # item is validated ItemCreate instance
    return {
        "id": 1,
        "created_at": datetime.now(timezone.utc),
        **item.model_dump()
    }

@app.patch("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int, item: ItemUpdate):
    # Only provided fields are included
    update_data = item.model_dump(exclude_unset=True)
    return {"id": item_id, "created_at": datetime.now(timezone.utc), **update_data}

Examples

Nested models:

class Address(BaseModel):
    street: str
    city: str
    postcode: str

class User(BaseModel):
    name: str
    email: EmailStr
    address: Address

@app.post("/users")
async def create_user(user: User):
    return user

Response with exclude/include:

class UserFull(BaseModel):
    id: int
    email: str
    password_hash: str
    is_active: bool

@app.get(
    "/users/{user_id}",
    response_model=UserFull,
    response_model_exclude={"password_hash"}
)
async def get_user(user_id: int):
    return {"id": user_id, "email": "user@example.com",
            "password_hash": "secret", "is_active": True}

Union types for multiple response models:

from typing import Union

class Cat(BaseModel):
    type: str = "cat"
    meow_volume: int

class Dog(BaseModel):
    type: str = "dog"
    bark_volume: int

@app.get("/pets/{pet_id}", response_model=Union[Cat, Dog])
async def get_pet(pet_id: int):
    if pet_id % 2 == 0:
        return Cat(meow_volume=5)
    return Dog(bark_volume=10)

Dependency Injection

FastAPI's dependency injection system enables clean, reusable, and testable code by managing shared resources and cross-cutting concerns.

⤢ zoom

Key Concepts

• Dependencies: Callable objects (functions or classes) that provide resources
• Depends: FastAPI's way to declare and inject dependencies
• Sub-dependencies: Dependencies can have their own dependencies
• Yield dependencies: Resources that need cleanup (e.g., database sessions)

Common Patterns

from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated

app = FastAPI()

# Simple dependency
def common_parameters(skip: int = 0, limit: int = 100):
    return {"skip": skip, "limit": limit}

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

# Database session dependency with cleanup
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

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

# Authentication dependency
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)]
):
    user = decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials"
        )
    return user

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

# Class-based dependency
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/products")
async def list_products(
    pagination: Annotated[Pagination, Depends()]
):
    return {"skip": pagination.skip, "limit": pagination.limit}

Examples

Chained dependencies:

def verify_token(token: str = Header(...)):
    if token != "valid-token":
        raise HTTPException(status_code=401)
    return token

def get_current_user(token: str = Depends(verify_token)):
    return {"username": "john", "token": token}

def get_current_admin(user: dict = Depends(get_current_user)):
    if user.get("role") != "admin":
        raise HTTPException(status_code=403)
    return user

@app.get("/admin")
async def admin_endpoint(admin: dict = Depends(get_current_admin)):
    return admin

Path operation level dependencies:

async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=403)

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

Global dependencies:

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

# Or on a router
router = APIRouter(dependencies=[Depends(get_current_user)])

Middleware and CORS

Middleware processes requests/responses globally, whilst CORS configuration enables cross-origin requests.

Key Concepts

• Middleware: Code that runs before/after every request
• CORS (Cross-Origin Resource Sharing): Security feature for browser-based requests
• Request/response lifecycle: Order of middleware execution
• Custom headers: Adding headers to all responses

Common Patterns

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    expose_headers=["X-Custom-Header"],
    max_age=600,  # Cache preflight requests for 10 minutes
)

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

# Class-based middleware
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Before request
        print(f"Request: {request.method} {request.url}")

        response = await call_next(request)

        # After request
        print(f"Response: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)

Examples

Authentication middleware:

from fastapi import HTTPException

@app.middleware("http")
async def authenticate_request(request: Request, call_next):
    # Skip authentication for certain paths
    if request.url.path in ["/health", "/docs", "/openapi.json"]:
        return await call_next(request)

    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"detail": "Missing or invalid token"}
        )

    return await call_next(request)

Request ID middleware:

import uuid
from fastapi.responses import JSONResponse

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

@app.get("/")
async def root(request: Request):
    return {"request_id": request.state.request_id}

CORS for development (allow all):

# Only use in development!
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Background Tasks

Background tasks run after the response is sent, useful for operations that don't need to block the response.

Key Concepts

• BackgroundTasks: FastAPI's built-in mechanism for async tasks
• Non-blocking: Tasks run after the response is returned to the client
• Use cases: Email notifications, logging, cache updates, cleanup
• Limitations: Not suitable for long-running tasks (use Celery/Redis Queue instead)

Common Patterns

from fastapi import FastAPI, BackgroundTasks
from typing import Annotated

app = FastAPI()

# Simple background task
def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/items")
async def create_item(
    item: dict,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, f"Item created: {item}")
    return item

# Async background task
async def send_email(email: str, subject: str, body: str):
    # Simulate email sending
    await asyncio.sleep(1)
    print(f"Email sent to {email}: {subject}")

@app.post("/register")
async def register_user(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(
        send_email,
        email,
        "Welcome!",
        "Thanks for registering"
    )
    return {"message": "Registration successful"}

# Multiple background tasks
@app.post("/orders")
async def create_order(
    order: dict,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_order_confirmation, order)
    background_tasks.add_task(update_inventory, order)
    background_tasks.add_task(notify_warehouse, order)
    return {"status": "Order placed"}

Examples

Background tasks in dependencies:

def log_request(
    background_tasks: BackgroundTasks,
    request: Request
):
    def write_log():
        with open("access.log", "a") as f:
            f.write(f"{request.method} {request.url}\n")

    background_tasks.add_task(write_log)

@app.get("/items", dependencies=[Depends(log_request)])
async def list_items():
    return []

Background task with database:

async def update_statistics(db: Session, item_id: int):
    # Update view count in database
    item = db.query(Item).filter(Item.id == item_id).first()
    if item:
        item.view_count += 1
        db.commit()

@app.get("/items/{item_id}")
async def get_item(
    item_id: int,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    item = db.query(Item).filter(Item.id == item_id).first()
    background_tasks.add_task(update_statistics, db, item_id)
    return item

Cleanup task:

import os
import tempfile

def cleanup_temp_file(filepath: str):
    if os.path.exists(filepath):
        os.remove(filepath)

@app.post("/upload")
async def upload_file(
    file: UploadFile,
    background_tasks: BackgroundTasks
):
    # Save temporarily (mktemp() is insecure — race condition)
    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(await file.read())
        temp_path = f.name

    # Process file...
    result = process_file(temp_path)

    # Schedule cleanup
    background_tasks.add_task(cleanup_temp_file, temp_path)

    return result

Testing and Documentation

FastAPI provides automatic interactive documentation and a TestClient for comprehensive testing.

Key Concepts

• Automatic docs: Swagger UI at /docs and ReDoc at /redoc
• OpenAPI schema: Auto-generated at /openapi.json
• TestClient: Synchronous testing client based on httpx (Starlette is migrating this to httpx2 — current versions emit a deprecation warning suggesting pip install httpx2)
• Dependency overrides: Replace dependencies during testing

Common Patterns

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
import pytest

app = FastAPI(
    title="My API",
    description="API description",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json"
)

# Endpoint with documentation
@app.get(
    "/items/{item_id}",
    summary="Get an item",
    description="Retrieve an item by its ID",
    response_description="The item details",
    responses={
        404: {"description": "Item not found"},
        500: {"description": "Internal server error"}
    }
)
async def get_item(item_id: int):
    """
    Get an item with all its details:

    - **item_id**: The unique identifier

    Returns the item or 404 if not found.
    """
    return {"item_id": item_id}

# Testing
client = TestClient(app)

def test_get_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_get_item_invalid():
    response = client.get("/items/invalid")
    assert response.status_code == 422  # Validation error

Examples

Testing with dependency overrides:

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

@app.get("/users")
async def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

# Test with override
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

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

Testing with authentication:

def test_protected_endpoint():
    # Without token
    response = client.get("/protected")
    assert response.status_code == 401

    # With valid token
    response = client.get(
        "/protected",
        headers={"Authorization": "Bearer valid-token"}
    )
    assert response.status_code == 200

def test_create_item():
    response = client.post(
        "/items",
        json={"name": "Widget", "price": 9.99}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Widget"
    assert "id" in data

Pytest fixtures:

import pytest
from fastapi.testclient import TestClient

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

@pytest.fixture
def authenticated_client(client):
    client.headers["Authorization"] = "Bearer test-token"
    return client

def test_with_auth(authenticated_client):
    response = authenticated_client.get("/me")
    assert response.status_code == 200

Async testing:

import pytest
from httpx import ASGITransport, AsyncClient

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

Custom OpenAPI documentation:

from fastapi.openapi.utils import get_openapi

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

    openapi_schema = get_openapi(
        title="Custom API",
        version="2.0.0",
        description="Custom description",
        routes=app.routes,
    )

    # Add custom info
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Quick Reference

Task	Code
Create app	app = FastAPI()
GET endpoint	@app.get("/path")
POST endpoint	@app.post("/path", status_code=201)
Path parameter	def func(item_id: int)
Query parameter	def func(skip: int = 0)
Request body	def func(item: ItemModel)
Response model	@app.get("/", response_model=Model)
Dependency	def func(db = Depends(get_db))
Background task	bg.add_task(func, arg1, arg2)
Add middleware	app.add_middleware(MiddlewareClass)
Include router	app.include_router(router)
Run server	uvicorn main:app --reload
Test client	client = TestClient(app)
Override dependency	app.dependency_overrides[dep] = override

Common Imports

from fastapi import (
    FastAPI, APIRouter, Depends, HTTPException,
    status, Query, Path, Body, Header, Cookie,
    BackgroundTasks, Request, Response, UploadFile
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
from typing import Optional, List, Annotated

CLI Commands

# Run development server with auto-reload
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Run production server with multiple workers
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000

# Generate OpenAPI schema
python -c "import json; from main import app; print(json.dumps(app.openapi()))"

Common Issues and Solutions

Issue	Solution
422 Validation Error	Check request body matches Pydantic model; verify Content-Type header is application/json
CORS errors in browser	Add CORSMiddleware with correct origins; ensure preflight (OPTIONS) requests are handled
Dependency not found	Ensure dependency is imported and Depends() is used correctly
Circular imports	Use from __future__ import annotations or lazy imports
Async/await errors	Use async def for async operations; don't mix sync and async incorrectly
Response model filtering	Set response_model_exclude_unset=True to exclude default values
Background task not running	Ensure task function is passed (not called): add_task(func, args) not add_task(func(args))
TestClient hanging	Use with TestClient(app) as client: context manager
Slow startup	Use a lifespan context manager for expensive initialisations (@app.on_event("startup") is deprecated)
Type hints not working	Upgrade to Python 3.9+ or use from __future__ import annotations

Debugging Tips

# Enable debug mode
import logging
logging.basicConfig(level=logging.DEBUG)

# Print all routes
for route in app.routes:
    print(f"{route.methods} {route.path}")

# Access request in exception handler
@app.exception_handler(Exception)
async def debug_exception_handler(request: Request, exc: Exception):
    import traceback
    return JSONResponse(
        status_code=500,
        content={
            "detail": str(exc),
            "traceback": traceback.format_exc()
        }
    )

Related Topics

The following topics complement FastAPI development and would make useful additions to your reference collection:

1. Python - Pydantic: Deep dive into data validation and settings management (already completed)
2. Python - SQLAlchemy: Database ORM commonly used with FastAPI for data persistence
3. Python - Redis: Caching and session storage for high-performance APIs
4. Python - Requests: HTTP client library for testing and external API calls
5. Python - Debugging Tools: Profiling and debugging FastAPI applications
6. API Design Patterns: RESTful principles, versioning, and authentication patterns