fastapi-expert

FastAPI Expert

Deep expertise in async Python, Pydantic V2, and production-grade API development with FastAPI.

When to Use This Skill

• Building REST APIs with FastAPI
• Implementing Pydantic V2 validation schemas
• Setting up async database operations
• Implementing JWT authentication/authorization
• Creating WebSocket endpoints
• Optimizing API performance

Core Workflow

1. Analyze requirements — Identify endpoints, data models, auth needs
2. Design schemas — Create Pydantic V2 models for validation
3. Implement — Write async endpoints with proper dependency injection
4. Secure — Add authentication, authorization, rate limiting
5. Test — Write async tests with pytest and httpx; run pytest after each endpoint group and verify OpenAPI docs at /docs

Checkpoint after each step: confirm schemas validate correctly, endpoints return expected HTTP status codes, and /docs reflects the intended API surface before proceeding.

Minimal Complete Example

Schema + endpoint + dependency injection in one cohesive unit:

# schemas.py
from pydantic import BaseModel, EmailStr, field_validator, model_config

class UserCreate(BaseModel):
    model_config = model_config(str_strip_whitespace=True)

    email: EmailStr
    password: str
    name: str | None = None

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

class UserResponse(BaseModel):
    model_config = model_config(from_attributes=True)

    id: int
    email: EmailStr
    name: str | None = None

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated

from app.database import get_db
from app.schemas import UserCreate, UserResponse
from app import crud

router = APIRouter(prefix="/users", tags=["users"])

DbDep = Annotated[AsyncSession, Depends(get_db)]

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: DbDep) -> UserResponse:
    existing = await crud.get_user_by_email(db, payload.email)
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
    return await crud.create_user(db, payload)