---

name: FastAPI Modern Web Development skill_id: fastapi-web-development version: 1.0.0 description: Production-grade FastAPI development with async patterns, Pydantic v2, dependency injection, ML/AI endpoint design, and modern Python best practices for building high-performance REST APIs category: Python Web Development tags:

• fastapi
• python
• async
• pydantic
• rest-api
• ml-endpoints
• dependency-injection
• web-framework
• async-programming
• api-design author: mcp-skillset license: MIT created: 2025-11-25 last_updated: 2025-11-25 toolchain:
• Python 3.11+
• FastAPI 0.100+
• Pydantic v2
• uvicorn
• httpx frameworks:
• FastAPI
• Pydantic
• SQLAlchemy 2.0
• Alembic related_skills:
• test-driven-development
• systematic-debugging
• postgresql-optimization
• security-testing

---

FastAPI Modern Web Development

Overview

This skill provides comprehensive guidance for building production-grade FastAPI applications with modern Python patterns (2024-2025 best practices). FastAPI is the #1 framework for AI/ML APIs, combining high performance, automatic OpenAPI documentation, and intuitive async/await patterns.

When to Use This Skill

Use this skill when:

• Building RESTful APIs for ML/AI services
• Creating high-performance async Python web services
• Developing data-intensive applications requiring concurrent request handling
• Implementing microservices with automatic API documentation
• Building APIs that require strong type safety and validation
• Designing endpoints for LLM integration and AI workflows

Core Principles

1. Async-First Architecture

Always prefer async/await for I/O-bound operations

Copyfrom fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
import httpx

app = FastAPI()

# CORRECT: Async database operations
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# CORRECT: Async external API calls
@app.get("/external-data")
async def fetch_external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

# WRONG: Blocking synchronous calls in async context
@app.get("/bad-example")
async def bad_handler():
    time.sleep(5)  # Blocks entire event loop!
    return {"status": "done"}

Why: FastAPI runs on ASGI (asyncio). Blocking calls prevent other requests from processing, degrading performance under load.

2. Pydantic v2 Models for Type Safety

Use Pydantic models for all request/response validation

Copyfrom pydantic import BaseModel, Field, field_validator, ConfigDict
from datetime import datetime
from typing import Optional

class UserCreate(BaseModel):
    """Request model for user creation"""
    model_config = ConfigDict(str_strip_whitespace=True)

    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: Optional[int] = Field(None, ge=13, le=120)

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v.lower()

class UserResponse(BaseModel):
    """Response model - never expose internal fields"""
    model_config = ConfigDict(from_attributes=True)  # Pydantic v2

    id: int
    username: str
    email: str
    created_at: datetime
    # DON'T expose: password_hash, internal_flags, etc.

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    db_user = User(**user.model_dump())  # Pydantic v2 syntax
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

Key Updates for Pydantic v2:

• Config → model_config = ConfigDict(...)
• orm_mode=True → from_attributes=True
• dict() → model_dump()
• @validator → @field_validator

3. Dependency Injection for Loose Coupling

Use FastAPI's dependency injection for reusable components

Copyfrom fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# Database session dependency
async_engine = create_async_engine("postgresql+asyncpg://...")
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# Authentication dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = verify_token(token)
    if payload is None:
        raise credentials_exception

    user = await db.get(User, payload.get("sub"))
    if user is None:
        raise credentials_exception
    return user

# Use dependencies in route
@app.get("/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

Benefits: Automatic injection, easy testing with override_dependency, shared logic across routes.

4. Structured Error Handling

Always use HTTPException with proper status codes

Copyfrom fastapi import HTTPException, status

@app.get("/items/{item_id}")
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
    item = await db.get(Item, item_id)

    # CORRECT: Specific HTTP exception
    if item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found"
        )

    if not item.is_public and not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient permissions"
        )

    return item

# Custom exception handler for domain errors
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"message": f"Item {exc.item_id} not found"}
    )

5. ML/AI Endpoint Design Patterns

Optimize endpoints for ML model serving

Copyfrom fastapi import BackgroundTasks
from functools import lru_cache
import asyncio

# Singleton pattern for model loading
@lru_cache()
def get_ml_model():
    """Load model once, cache for all requests"""
    import torch
    model = torch.load("model.pth")
    model.eval()
    return model

class PredictionRequest(BaseModel):
    text: str = Field(..., max_length=10000)
    temperature: float = Field(0.7, ge=0.0, le=2.0)

class PredictionResponse(BaseModel):
    prediction: str
    confidence: float
    processing_time_ms: float

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    model = get_ml_model()

    start_time = asyncio.get_event_loop().time()

    # Run CPU-intensive model inference in thread pool
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,  # Uses default ThreadPoolExecutor
        model.predict,
        request.text
    )

    elapsed = (asyncio.get_event_loop().time() - start_time) * 1000

    return PredictionResponse(
        prediction=result["text"],
        confidence=result["score"],
        processing_time_ms=elapsed
    )

# Background task for async processing
@app.post("/predict-async")
async def predict_async(
    request: PredictionRequest,
    background_tasks: BackgroundTasks
):
    task_id = generate_task_id()
    background_tasks.add_task(process_prediction, task_id, request)
    return {"task_id": task_id, "status": "processing"}

Best Practices

Application Structure

Copyfastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app, startup/shutdown events
│   ├── config.py            # Pydantic Settings management
│   ├── models/              # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic models (request/response)
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── api/                 # API routes
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── items.py
│   │   └── deps.py          # Shared dependencies
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── ml_service.py
│   ├── core/                # Core utilities
│   │   ├── __init__.py
│   │   ├── security.py
│   │   └── database.py
│   └── tests/
│       ├── __init__.py
│       ├── conftest.py
│       └── test_users.py
├── alembic/                 # Database migrations
├── .env
├── pyproject.toml
└── README.md

Configuration Management

Copyfrom pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False
    )

    # Database
    DATABASE_URL: str = "postgresql+asyncpg://localhost/dbname"

    # Security
    SECRET_KEY: str
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # API
    API_V1_PREFIX: str = "/api/v1"
    PROJECT_NAME: str = "My FastAPI Project"

    # CORS
    BACKEND_CORS_ORIGINS: list[str] = ["http://localhost:3000"]

@lru_cache()
def get_settings() -> Settings:
    return Settings()

# Use in dependencies
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
    return {"project_name": settings.PROJECT_NAME}

Request Lifecycle & Middleware

Copyfrom fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time
import logging

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Compression middleware
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom logging middleware
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    logging.info(
        f"{request.method} {request.url.path} "
        f"completed in {process_time:.3f}s "
        f"with status {response.status_code}"
    )

    response.headers["X-Process-Time"] = str(process_time)
    return response

Common Patterns

Pagination

Copyfrom typing import Generic, TypeVar, Sequence
from pydantic import BaseModel

T = TypeVar('T')

class Page(BaseModel, Generic[T]):
    items: Sequence[T]
    total: int
    page: int
    size: int
    pages: int

async def paginate(
    query,
    page: int = 1,
    size: int = 50
) -> Page[T]:
    total = await db.scalar(select(func.count()).select_from(query))
    items = await db.scalars(
        query.offset((page - 1) * size).limit(size)
    )

    return Page(
        items=items.all(),
        total=total,
        page=page,
        size=size,
        pages=(total + size - 1) // size
    )

@app.get("/users", response_model=Page[UserResponse])
async def list_users(
    page: int = 1,
    size: int = 50,
    db: AsyncSession = Depends(get_db)
):
    query = select(User).order_by(User.created_at.desc())
    return await paginate(query, page, size)

File Upload with Validation

Copyfrom fastapi import File, UploadFile
from PIL import Image
import aiofiles

@app.post("/upload-image")
async def upload_image(file: UploadFile = File(...)):
    # Validate file type
    if not file.content_type.startswith("image/"):
        raise HTTPException(400, "File must be an image")

    # Validate file size (10MB limit)
    contents = await file.read()
    if len(contents) > 10 * 1024 * 1024:
        raise HTTPException(400, "File too large (max 10MB)")

    # Validate image can be opened
    try:
        image = Image.open(BytesIO(contents))
        image.verify()
    except Exception:
        raise HTTPException(400, "Invalid image file")

    # Save file asynchronously
    file_path = f"uploads/{file.filename}"
    async with aiofiles.open(file_path, 'wb') as f:
        await f.write(contents)

    return {"filename": file.filename, "size": len(contents)}

WebSocket for Real-Time Updates

Copyfrom 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")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Anti-Patterns to Avoid

❌ DON'T: Mix sync and async incorrectly

Copy# WRONG: Sync function in async route
@app.get("/bad")
async def bad_endpoint():
    result = sync_database_call()  # Blocks event loop!
    return result

# CORRECT: Use run_in_executor for sync calls
@app.get("/good")
async def good_endpoint():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, sync_database_call)
    return result

❌ DON'T: Return ORM models directly

Copy# WRONG: Exposes internal fields, risks lazy loading
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    return await db.get(User, user_id)  # Returns ORM model!

# CORRECT: Use Pydantic response model
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return user  # FastAPI converts to UserResponse

❌ DON'T: Hardcode configuration

Copy# WRONG: Hardcoded values
DATABASE_URL = "postgresql://localhost/db"

# CORRECT: Use Pydantic Settings
settings = get_settings()
DATABASE_URL = settings.DATABASE_URL

❌ DON'T: Ignore startup/shutdown events

Copy# CORRECT: Cleanup resources
@app.on_event("startup")
async def startup():
    # Initialize database pool, load ML models, etc.
    app.state.db = await create_db_pool()

@app.on_event("shutdown")
async def shutdown():
    # Close connections, cleanup
    await app.state.db.close()

Testing Strategy

Copyfrom fastapi.testclient import TestClient
from httpx import AsyncClient
import pytest

# Sync tests with TestClient
def test_create_user():
    client = TestClient(app)
    response = client.post(
        "/users",
        json={"username": "testuser", "email": "test@example.com"}
    )
    assert response.status_code == 201
    assert response.json()["username"] == "testuser"

# Async tests with httpx
@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/1")
        assert response.status_code == 200

# Override dependencies for testing
def override_get_db():
    return test_database_session

app.dependency_overrides[get_db] = override_get_db

Performance Optimization

1. Connection Pooling: Use SQLAlchemy async engine with pool_size=20, max_overflow=0
2. Response Caching: Use Redis with @lru_cache() for expensive computations
3. Background Tasks: Offload non-critical work to BackgroundTasks
4. Async Libraries: Use httpx (not requests), asyncpg (not psycopg2)
5. Profiling: Use uvicorn --log-level debug and timing middleware

Security Checklist

• ✅ Use HTTPS in production (via reverse proxy)
• ✅ Implement rate limiting (slowapi library)
• ✅ Validate all inputs with Pydantic models
• ✅ Use OAuth2 with JWT for authentication
• ✅ Never expose internal error details in production
• ✅ Set proper CORS origins (not ["*"] in production)
• ✅ Hash passwords with bcrypt or argon2
• ✅ Use SECRET_KEY from environment, never hardcode

Related Skills

• test-driven-development: Write tests before implementing endpoints
• systematic-debugging: Debug async issues and race conditions
• postgresql-optimization: Optimize database queries for FastAPI
• security-testing: Perform security audits on API endpoints

Additional Resources

• FastAPI Official Docs: https://fastapi.tiangolo.com
• Pydantic v2 Migration: https://docs.pydantic.dev/latest/migration/
• SQLAlchemy 2.0 Async: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
• Production Deployment: https://fastapi.tiangolo.com/deployment/

Example Questions to Ask

• "How do I implement JWT authentication in FastAPI?"
• "What's the best way to handle file uploads asynchronously?"
• "How do I optimize this endpoint for ML model inference?"
• "Show me how to add pagination to this query"
• "How do I write async tests for FastAPI endpoints?"
• "What's the correct way to handle database transactions?"