Fastapi
FastAPI patterns for async APIs, dependency injection, Pydantic models, and OpenAPI integration
You are an expert in FastAPI for building high-performance async APIs with automatic OpenAPI documentation. ## Key Points - Use `response_model` on every endpoint to control what gets serialized and to generate accurate OpenAPI docs. - Structure large projects with `APIRouter` per domain area and include them in the main app with prefixes. - Use `Depends()` for all shared logic (database sessions, auth, feature flags) rather than importing singletons. - Use `from_attributes=True` (Pydantic v2) on response models to serialize ORM objects directly. - Run with `--workers` matching your CPU count for sync workloads, or use a single worker with async for I/O-bound workloads. - Add request ID middleware to correlate logs across services. - Mixing sync and async database drivers. Using synchronous `psycopg2` inside an async endpoint blocks the event loop. Use `asyncpg` or `psycopg[async]`. - Forgetting to `await` coroutines. FastAPI will silently return a coroutine object instead of raising an error if you forget `await` inside an endpoint. - Defining dependencies at module scope instead of using `Depends()`. This breaks testability and creates hidden global state. - Returning ORM objects directly without a `response_model`. FastAPI falls back to `jsonable_encoder`, which may serialize fields you did not intend to expose. - Not handling `lifespan` properly. Startup/shutdown logic must use the `lifespan` context manager; the old `on_event` decorators are deprecated. ## Quick Example ```bash pip install "fastapi[standard]" uvicorn[standard] sqlalchemy[asyncio] asyncpg ``` ```bash uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 ```
skilldb get python-web-skills/FastapiFull skill: 274 linesFastAPI — Python Web Development
You are an expert in FastAPI for building high-performance async APIs with automatic OpenAPI documentation.
Core Philosophy
Overview
FastAPI is a modern Python web framework built on Starlette and Pydantic. It provides automatic request validation, serialization, dependency injection, and interactive API docs. Its async-first design makes it well suited for I/O-bound services.
Setup & Configuration
pip install "fastapi[standard]" uvicorn[standard] sqlalchemy[asyncio] asyncpg
# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from .database import engine
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Shutdown
await engine.dispose()
app = FastAPI(
title="My API",
version="1.0.0",
lifespan=lifespan,
)
Run with:
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
Core Patterns
Pydantic Schemas
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
class ArticleCreate(BaseModel):
title: str = Field(..., min_length=5, max_length=200)
body: str
tag_ids: list[int] = []
class ArticleResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str
body: str
author_id: int
published_at: datetime | None
created_at: datetime
class PaginatedResponse(BaseModel):
items: list[ArticleResponse]
total: int
page: int
page_size: int
Path Operations
from fastapi import APIRouter, HTTPException, Query, Path, status
router = APIRouter(prefix="/articles", tags=["articles"])
@router.get("/", response_model=PaginatedResponse)
async def list_articles(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
search: str | None = Query(None),
db: AsyncSession = Depends(get_db),
):
query = select(Article)
if search:
query = query.where(Article.title.ilike(f"%{search}%"))
total = await db.scalar(select(func.count()).select_from(query.subquery()))
results = await db.scalars(
query.offset((page - 1) * page_size).limit(page_size)
)
return PaginatedResponse(
items=results.all(),
total=total,
page=page,
page_size=page_size,
)
@router.post("/", response_model=ArticleResponse, status_code=status.HTTP_201_CREATED)
async def create_article(
data: ArticleCreate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
article = Article(**data.model_dump(), author_id=current_user.id)
db.add(article)
await db.commit()
await db.refresh(article)
return article
@router.get("/{article_id}", response_model=ArticleResponse)
async def get_article(
article_id: int = Path(..., gt=0),
db: AsyncSession = Depends(get_db),
):
article = await db.get(Article, article_id)
if not article:
raise HTTPException(status_code=404, detail="Article not found")
return article
Dependency Injection
from fastapi import Depends, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
security = HTTPBearer()
async def get_db():
async with async_session_factory() as session:
yield session
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Security(security),
db: AsyncSession = Depends(get_db),
) -> User:
token = credentials.credentials
payload = decode_jwt(token)
user = await db.get(User, payload["sub"])
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
def require_role(role: str):
async def check_role(user: User = Depends(get_current_user)):
if role not in user.roles:
raise HTTPException(status_code=403, detail="Insufficient permissions")
return user
return check_role
Middleware and Exception Handling
import time
from fastapi import Request
from fastapi.responses import JSONResponse
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
elapsed = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{elapsed:.4f}"
return response
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(
status_code=400,
content={"detail": str(exc)},
)
Background Tasks
from fastapi import BackgroundTasks
async def send_notification(email: str, article_id: int):
# send email logic
pass
@router.post("/{article_id}/publish")
async def publish_article(
article_id: int,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
article = await db.get(Article, article_id)
article.publish()
await db.commit()
background_tasks.add_task(send_notification, user.email, article_id)
return {"status": "published"}
Testing
import pytest
from httpx import AsyncClient, ASGITransport
from main import app
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
@pytest.mark.anyio
async def test_list_articles(client: AsyncClient):
response = await client.get("/articles/")
assert response.status_code == 200
data = response.json()
assert "items" in data
Best Practices
- Use
response_modelon every endpoint to control what gets serialized and to generate accurate OpenAPI docs. - Structure large projects with
APIRouterper domain area and include them in the main app with prefixes. - Use
Depends()for all shared logic (database sessions, auth, feature flags) rather than importing singletons. - Use
from_attributes=True(Pydantic v2) on response models to serialize ORM objects directly. - Run with
--workersmatching your CPU count for sync workloads, or use a single worker with async for I/O-bound workloads. - Add request ID middleware to correlate logs across services.
Common Pitfalls
- Mixing sync and async database drivers. Using synchronous
psycopg2inside an async endpoint blocks the event loop. Useasyncpgorpsycopg[async]. - Forgetting to
awaitcoroutines. FastAPI will silently return a coroutine object instead of raising an error if you forgetawaitinside an endpoint. - Defining dependencies at module scope instead of using
Depends(). This breaks testability and creates hidden global state. - Returning ORM objects directly without a
response_model. FastAPI falls back tojsonable_encoder, which may serialize fields you did not intend to expose. - Not handling
lifespanproperly. Startup/shutdown logic must use thelifespancontext manager; the oldon_eventdecorators are deprecated.
Anti-Patterns
Over-engineering for hypothetical scale. Building for millions of users when you have hundreds adds complexity without value. Solve today's problems first.
Ignoring the existing ecosystem. Reinventing functionality that mature libraries already provide well wastes time and introduces unnecessary risk.
Premature abstraction. Creating elaborate frameworks and utilities before you have enough concrete cases to know what the abstraction should look like produces the wrong abstraction.
Neglecting error handling at boundaries. Internal code can trust its inputs, but system boundaries (user input, APIs, file I/O) require defensive validation.
Skipping documentation for obvious code. What is obvious to you today will not be obvious to your colleague next month or to you next year.
Install this skill directly: skilldb add python-web-skills
Related Skills
Celery
Celery patterns for distributed task queues, scheduling, retries, and worker management
Django Admin
Django admin customization patterns for list views, forms, inlines, actions, and permissions
Django ORM
Django ORM patterns for models, querysets, migrations, and database optimization
Django REST Framework
Django REST Framework patterns for building, serializing, and securing RESTful APIs
Flask
Flask application patterns for routing, blueprints, extensions, and application factories
Python Websockets
WebSocket patterns for real-time communication using FastAPI, Django Channels, and the websockets library