fastapi-patterns

Production FastAPI patterns with Pydantic v2, async endpoints, OAuth2/JWT, dependency injection, testing, and Kubernetes deployment

Safety Notice

This listing is from the official public ClawHub registry. Review SKILL.md and referenced scripts before running.

Copy this and send it to your AI assistant to learn

Install skill "fastapi-patterns" with this command: npx skills add msruruguay/fastapi-patterns

FastAPI Production Patterns

Description

Modern FastAPI development patterns for 2026, covering the full spectrum from Pydantic v2 migration to production deployment. Validated against 15+ production sources with 0.95 confidence. This skill keeps you on the current side of breaking changes and equips you with battle-tested patterns for async, security, testing, and performance.

Usage

Install this skill to get production-ready FastAPI patterns including:

  • Pydantic v2 migration guide (breaking changes, 4-17x performance gains)
  • Async vs sync endpoint decision rules
  • OAuth2 + JWT authentication patterns
  • Dependency override testing strategies
  • Docker + Kubernetes deployment with health probes

When working on FastAPI projects, this skill provides context for:

  • Migrating from Pydantic v1 to v2 without breaking existing code
  • Setting up proper lifespan events instead of deprecated @app.on_event()
  • Structuring tests with TestClient (sync) and AsyncClient (async)
  • Configuring Gunicorn + Uvicorn workers for production

Key Patterns

Pydantic v2 Migration

# BEFORE (Pydantic v1 — deprecated)
from pydantic import BaseModel, validator

class User(BaseModel):
    name: str

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v:
            raise ValueError('name cannot be empty')
        return v

# AFTER (Pydantic v2 — current)
from pydantic import BaseModel, field_validator

class User(BaseModel):
    name: str

    @field_validator('name')
    @classmethod
    def name_must_not_be_empty(cls, v):
        if not v:
            raise ValueError('name cannot be empty')
        return v

Method renames (v1 to v2): parse_obj() -> model_validate(), dict() -> model_dump(), json() -> model_dump_json(), orm_mode = True -> model_config = ConfigDict(from_attributes=True), conint(ge=0) -> Annotated[int, Field(ge=0)]

Lifespan Events (Modern Pattern)

from contextlib import asynccontextmanager
from fastapi import FastAPI

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load resources
    ml_models["answer"] = load_ml_model()
    yield
    # Shutdown: release resources
    ml_models.clear()

app = FastAPI(lifespan=lifespan)

Do NOT use deprecated @app.on_event(). Use lifespan for: DB connection pools, ML models, resource cleanup.

Async Decision Rule

# Use async def for: asyncpg, motor, httpx (non-blocking I/O)
@app.get('/users/')
async def get_users():
    users = await async_db.fetch('SELECT * FROM users')
    return users

# Use def for: psycopg2, pymongo, requests (blocking libs)
# FastAPI automatically runs def endpoints in a threadpool
@app.get('/users/sync')
def get_users_sync():
    return sync_db.query('SELECT * FROM users')

Rule of thumb: If unsure, use def. FastAPI handles the threadpool for you.

OAuth2 + JWT Authentication

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise HTTPException(status_code=401)
        return username
    except:
        raise HTTPException(status_code=401)

CORS (Never Use wildcard in Production)

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # NOT ["*"]
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Async Testing with httpx

import pytest
from httpx import ASGITransport, AsyncClient

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

Dependency Overrides for Mocking

async def override_get_db():
    db = Database("sqlite:///:memory:")
    yield db
    await db.disconnect()

def test_with_mock_db():
    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    response = client.get("/users/")
    assert response.status_code == 200
    app.dependency_overrides = {}  # Always clean up

Production Dockerfile

FROM python:3.11-slim
WORKDIR /code
COPY ./requirements.txt /code/requirements.txt
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
COPY ./app /code/app
CMD ["gunicorn", "app.main:app", "--workers", "4",
     "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:80"]

Workers: 1 per CPU core on a single server. For Kubernetes: 1 worker per container, scale at cluster level.

Health Check Endpoints

@app.get("/health")
async def health_check():
    """Liveness: is the process running?"""
    return {"status": "healthy"}

@app.get("/ready")
async def readiness_check():
    """Readiness: can it handle traffic?"""
    try:
        await database.execute("SELECT 1")
        return {"status": "ready", "database": "ok"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))

Performance Quick Wins

# Connection pooling: 2-3x throughput improvement
database = Database("postgresql://user:pass@localhost/db", min_size=5, max_size=20)

# Response caching: 90%+ database load reduction
@cache(expire=60)
async def get_user(user_id: int):
    return await db.fetch_one(...)

# Async middleware: 10-30% latency reduction
@app.middleware("http")
async def async_middleware(request, call_next):
    return await call_next(request)  # NOT blocking def

Common Pitfall: Blocking Inside Async

# WRONG: blocks the event loop
@app.get("/data")
async def get_data():
    response = requests.get("https://api.com")  # Blocks!

# CORRECT: use async library
import httpx
@app.get("/data")
async def get_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.com")

Tools & References


Published by MidOS — MCP Community Library

Source Transparency

This detail page is rendered from real SKILL.md content. Trust labels are metadata-based hints, not a safety guarantee.

Related Skills

Related by shared tags or category signals.

Coding

FastAPI Production Engineering

Build, deploy, and scale production-grade FastAPI apps with async, type-safe design, structured errors, JWT auth, async ORM, auto docs, tests, and observabil...

Registry SourceRecently Updated
0326
Profile unavailable
Coding

joinquant

聚宽量化交易平台 - 提供A股、期货、基金数据查询,事件驱动策略回测,支持在线研究与模拟实盘。

Registry SourceRecently Updated
061
Profile unavailable
Coding

错敏信息检测

基于FastAPI的文本错敏信息检测服务,识别敏感词、错别字及规范表述问题,提供RESTful API接口调用。

Registry SourceRecently Updated
0177
Profile unavailable