fastapi-expert

name	fastapi-expert
description	Deep FastAPI expertise for async API development, dependency injection, Pydantic models, middleware, and testing. Use when building Python APIs with FastAPI, designing API schemas, or implementing authentication. Use when this capability is needed.
metadata	{"author":"bendourthe"}

FastAPI Expert

Specialized expertise in FastAPI development, providing deep guidance on async API design, Pydantic v2 model patterns, dependency injection, middleware, background tasks, WebSocket support, async database integration, testing strategies, OpenAPI customization, and production deployment.

When to Use This Skill

Use this skill for:

• Designing RESTful APIs with path, query, and body parameters
• Building Pydantic v2 models with validators, computed fields, and serialization
• Implementing dependency injection hierarchies and scoped dependencies
• Writing middleware for CORS, authentication, rate limiting, and logging
• Running background tasks and WebSocket connections
• Integrating async databases (SQLAlchemy async, Tortoise ORM)
• Testing with TestClient, async fixtures, and dependency overrides
• Customizing OpenAPI schema and documentation
• Deploying with uvicorn, gunicorn, and Docker

Trigger phrases: "fastapi", "fast api", "pydantic", "python api", "async api", "uvicorn", "python rest api", "fastapi dependency injection", "fastapi middleware", "fastapi testing"

What This Skill Does

Provides FastAPI expertise including:

• Route Design: Path parameters, query parameters, request body, response models
• Pydantic v2: Model validators, computed fields, serialization aliases, discriminated unions
• Dependency Injection: Sub-dependencies, yield dependencies, scoped lifetimes
• Middleware: CORS, authentication, rate limiting, request timing
• Background Tasks: Async background work, task queues
• WebSockets: Real-time connections with connection management
• Database: SQLAlchemy async sessions, repository pattern, migrations
• Testing: TestClient, httpx async client, dependency overrides, factory fixtures
• Deployment: uvicorn, gunicorn workers, Docker, health checks

Instructions

Step 1: Structure a FastAPI Project

Recommended project layout:

src/
  app/
    __init__.py
    main.py               # Application factory
    config.py             # Settings with Pydantic
    dependencies.py       # Shared dependencies
    middleware.py          # Custom middleware
    models/
      __init__.py
      user.py             # SQLAlchemy models
      post.py
    schemas/
      __init__.py
      user.py             # Pydantic request/response schemas
      post.py
      common.py           # Shared schemas (pagination, errors)
    routers/
      __init__.py
      users.py            # /users endpoints
      posts.py            # /posts endpoints
      auth.py             # /auth endpoints
    services/
      __init__.py
      user_service.py     # Business logic
      auth_service.py
    db/
      __init__.py
      session.py          # Database session management
      base.py             # SQLAlchemy Base
tests/
  conftest.py             # Shared fixtures
  test_users.py
  test_auth.py
  factories.py            # Test data factories

Application factory:

# src/app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.db.session import engine
from app.db.base import Base
from app.middleware import TimingMiddleware
from app.routers import users, posts, auth


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown events."""
    # Startup: create tables (use Alembic in production)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: dispose engine
    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        version="1.0.0",
        docs_url="/docs" if settings.debug else None,
        redoc_url="/redoc" if settings.debug else None,
        lifespan=lifespan,
    )

    # Middleware (order matters: last added = first executed)
    app.add_middleware(TimingMiddleware)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=,
        allow_methods=[],
        allow_headers=[],
    )

    
    app.include_router(auth.router, prefix=, tags=[])
    app.include_router(users.router, prefix=, tags=[])
    app.include_router(posts.router, prefix=, tags=[])

     app


app = create_app()

Settings with Pydantic:

# src/app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


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

    app_name: str = "My API"
    debug: bool = False
    database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/mydb"
    secret_key: str = "change-me-in-production"
    access_token_expire_minutes: int = 30
    cors_origins: list[str] = ["http://localhost:3000"]


settings = Settings()

Step 2: Design Pydantic v2 Schemas

# src/app/schemas/user.py
from datetime import datetime
from typing import Annotated

from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, computed_field


class UserBase(BaseModel):
    """Shared fields for user schemas."""
    email: EmailStr
    display_name: Annotated[str, Field(min_length=2, max_length=50)]


class UserCreate(UserBase):
    """Schema for creating a user."""
    password: Annotated[str, Field(min_length=8, max_length=128)]

    @field_validator("password")
    @classmethod
    def password_complexity(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain at least one digit")
        return v


class UserUpdate(BaseModel):
    
    display_name:  |  = 
    email: EmailStr |  = 


 ():
    
    model_config = ConfigDict(from_attributes=)

    : 
    created_at: datetime
    is_active: 



     () -> :
         (datetime.utcnow() - .created_at).days


 ():
    
    data: [UserResponse]
    total: 
    page: 
    page_size: 
    has_next: 

Discriminated union for polymorphic responses:

from typing import Literal
from pydantic import BaseModel


class TextNotification(BaseModel):
    type: Literal["text"] = "text"
    message: str


class ImageNotification(BaseModel):
    type: Literal["image"] = "image"
    image_url: str
    caption: str | None = None


class ActionNotification(BaseModel):
    type: Literal["action"] = "action"
    message: str
    action_url: str
    action_label: str


# FastAPI automatically generates correct OpenAPI schema
Notification = TextNotification | ImageNotification | ActionNotification

Step 3: Implement Dependency Injection

# src/app/dependencies.py
from typing import Annotated, AsyncGenerator

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.db.session import async_session_factory
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Yield a database session and ensure it is closed."""
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


# Type alias for cleaner signatures
DbSession = Annotated[AsyncSession, Depends(get_db)]


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DbSession,
) -> User:
    """Decode JWT and return the authenticated user."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": },
    )
    :
        payload = jwt.decode(token, settings.secret_key, algorithms=[])
        user_id:  |  = payload.get()
         user_id  :
             credentials_exception
     JWTError:
         credentials_exception

    user =  db.get(User, user_id)
     user     user.is_active:
         credentials_exception
     user


CurrentUser = Annotated[User, Depends(get_current_user)]


  () -> User:
    
     user.role != :
         HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=,
        )
     user


AdminUser = Annotated[User, Depends(require_admin)]

Step 4: Build Route Handlers

# src/app/routers/users.py
from fastapi import APIRouter, HTTPException, Query, status

from app.dependencies import CurrentUser, AdminUser, DbSession
from app.schemas.user import UserCreate, UserUpdate, UserResponse, UserListResponse
from app.services.user_service import UserService

router = APIRouter()


@router.get("", response_model=UserListResponse)
async def list_users(
    db: DbSession,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
    search: str | None = Query(default=None, max_length=100),
):
    """List users with pagination and optional search."""
    service = UserService(db)
    users, total = await service.list_users(
        page=page, page_size=page_size, search=search
    )
    return UserListResponse(
        data=users,
        total=total,
        page=page,
        page_size=page_size,
        has_next=(page * page_size) < total,
    )


@router.get("/me", response_model=UserResponse)
async def get_current_user_profile(user: CurrentUser):
    """Get the authenticated user's profile."""
    return user


@router.get(, response_model=UserResponse)
  ():
    
    service = UserService(db)
    user =  service.get_by_id(user_id)
      user:
         HTTPException(status_code=, detail=)
     user



  ():
    
    service = UserService(db)
    existing =  service.get_by_email(data.email)
     existing:
         HTTPException(status_code=, detail=)
      service.create(data)



  ():
    
     user. != user_id  user.role != :
         HTTPException(status_code=, detail=)
    service = UserService(db)
    updated =  service.update(user_id, data)
      updated:
         HTTPException(status_code=, detail=)
     updated



  ():
    
    service = UserService(db)
    deleted =  service.delete(user_id)
      deleted:
         HTTPException(status_code=, detail=)

Step 5: Middleware Patterns

# src/app/middleware.py
import time
import logging
from collections import defaultdict

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger(__name__)


class TimingMiddleware(BaseHTTPMiddleware):
    """Log request duration for every endpoint."""

    async def dispatch(self, request: Request, call_next) -> Response:
        start = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000
        response.headers["X-Process-Time-Ms"] = f"{duration_ms:.1f}"
        logger.info(
            "%s %s completed in %.1fms (status %d)",
            request.method,
            request.url.path,
            duration_ms,
            response.status_code,
        )
        return response


class RateLimitMiddleware(BaseHTTPMiddleware):
    """Simple in-memory rate limiter (use Redis in production)."""

    def __init__(self, app, requests_per_minute: int = 60):
        super().__init__(app)
        self.rpm = requests_per_minute
        self.requests: dict[str, list[float]] = defaultdict()

      () -> Response:
        client_ip = request.client.host  request.client  
        now = time.time()
        window_start = now - 

        
        .requests[client_ip] = [
            t  t  .requests[client_ip]  t > window_start
        ]

         (.requests[client_ip]) >= .rpm:
             Response(
                content=,
                status_code=,
                media_type=,
                headers={: },
            )

        .requests[client_ip].append(now)
          call_next(request)

Step 6: Database Integration (SQLAlchemy Async)

# src/app/db/session.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# src/app/models/user.py
from datetime import datetime

from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    display_name: Mapped[str] = mapped_column(String(50))
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(20), default="user")
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

# src/app/services/user_service.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.services.auth_service import hash_password


class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        return await self.db.get(User, user_id)

    async def get_by_email(self, email: str) -> User | None:
        result = await self.db.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def list_users(
        self, page: int, page_size: int, search: str | None
    ) -> tuple[list[User], int]:
        query = select(User).where(User.is_active.is_(True))
        count_query = select(func.count()).select_from(User).where(
            User.is_active.is_()
        )

         search:
            pattern = 
            query = query.where(
                User.display_name.ilike(pattern) | User.email.ilike(pattern)
            )
            count_query = count_query.where(
                User.display_name.ilike(pattern) | User.email.ilike(pattern)
            )

        total = ( .db.execute(count_query)).scalar()  
        offset = (page - ) * page_size
        result =  .db.execute(
            query.offset(offset).limit(page_size).order_by(User.created_at.desc())
        )
         (result.scalars().()), total

      () -> User:
        user = User(
            email=data.email,
            display_name=data.display_name,
            hashed_password=hash_password(data.password),
        )
        .db.add(user)
         .db.flush()
         .db.refresh(user)
         user

      () -> User | :
        user =  .get_by_id(user_id)
          user:
             
        update_data = data.model_dump(exclude_unset=)
         field, value  update_data.items():
            (user, field, value)
         .db.flush()
         .db.refresh(user)
         user

      () -> :
        user =  .get_by_id(user_id)
          user:
             
         .db.delete(user)
         

Step 7: Background Tasks and WebSockets

Background tasks:

from fastapi import BackgroundTasks

async def send_welcome_email(email: str, name: str) -> None:
    """Simulate sending an email (replace with real email service)."""
    # await email_client.send(to=email, subject="Welcome!", body=f"Hello {name}")
    logger.info("Welcome email sent to %s", email)


@router.post("/register", status_code=201)
async def register(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    db: DbSession,
):
    service = UserService(db)
    user = await service.create(data)
    background_tasks.add_task(send_welcome_email, user.email, user.display_name)
    return UserResponse.model_validate(user)

WebSocket endpoint:

from fastapi import WebSocket, WebSocketDisconnect


class ConnectionManager:
    """Manage active WebSocket connections."""

    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active:
            await connection.send_text(message)


manager = ConnectionManager()


@router.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast()

Step 8: Testing

Shared fixtures:

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.db.base import Base
from app.dependencies import get_db
from app.main import create_app

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@pytest_asyncio.fixture(autouse=True)
async def setup_db():
    """Create tables before each test, drop after."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest_asyncio.fixture
async def db_session() -> AsyncSession:
    async with TestSessionLocal() as session:
        yield session


@pytest_asyncio.fixture
async def client(db_session: AsyncSession):
    
    app = create_app()

      ():
         db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
      AsyncClient(transport=transport, base_url=)  ac:
         ac
    app.dependency_overrides.clear()

Route tests:

# tests/test_users.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users", json={
        "email": "[e-mail]",
        "display_name": "Test User",
        "password": "Str0ngPass!",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[e-mail]"
    assert data["display_name"] == "Test User"
    assert "hashed_password" not in data     # Verify password is excluded
    assert "id" in data


@pytest.mark.asyncio
async def test_create_user_duplicate_email(client: AsyncClient):
    payload = {
        "email": "[e-mail]",
        "display_name": "User One",
        "password": "Str0ngPass!",
    }
    await client.post("/users", json=payload)
    response = await client.post("/users", json={**payload, : })
     response.status_code == 



  ():
    response =  client.post(, json={
        : ,
        : ,
        : ,
    })
     response.status_code ==        



  ():
    
     i  ():
         client.post(, json={
            : ,
            : ,
            : ,
        })

    response =  client.get(, params={: , : })
     response.status_code == 
    data = response.json()
     (data[]) == 
     data[] == 
     data[]  

Step 9: Deployment

uvicorn with gunicorn (production):

# Direct uvicorn (development)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# gunicorn with uvicorn workers (production)
gunicorn app.main:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --access-logfile -

Dockerfile:

FROM python:3.12-slim AS base

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

FROM base AS deps
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev

FROM base AS runner
COPY --from=deps /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"

COPY src/ ./src/

# Non-root user
RUN addgroup --system --gid 1001 appgroup && \
    adduser --system --uid 1001 --ingroup appgroup appuser
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["gunicorn", "app.main:app", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--workers", "4", \
     "--bind", "0.0.0.0:8000"]

Health check endpoint:

@app.get("/health", include_in_schema=False)
async def health_check(db: DbSession):
    """Health check that verifies database connectivity."""
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "healthy", "database": "connected"}
    except Exception:
        return Response(
            content='{"status":"unhealthy","database":"disconnected"}',
            status_code=503,
            media_type="application/json",
        )

Best Practices

• Use async endpoints for I/O-bound operations: database queries, HTTP calls, file I/O
• Define response models explicitly: prevents leaking internal fields (e.g., hashed_password)
• Use Annotated type aliases for dependencies: DbSession, CurrentUser keep signatures clean
• Validate all input with Pydantic schemas: never trust raw request data
• Use dependency injection, not global state: makes testing trivial with overrides
• Commit in the dependency, not the route: the get_db dependency should commit on success and rollback on error
• Keep routes thin: delegate business logic to service classes
• Use structured logging: include request ID, user ID, and duration in log entries
• Pin dependency versions: use uv.lock or pip-compile for reproducible builds

Common Patterns

Pattern 1: Repository Pattern with Service Layer

# Separate data access (repository) from business logic (service)
class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def find_by_id(self, user_id: int) -> User | None:
        return await self.db.get(User, user_id)

    async def find_by_email(self, email: str) -> User | None:
        result = await self.db.execute(select(User).where(User.email == email))
        return result.scalar_one_or_none()

    async def save(self, user: User) -> User:
        self.db.add(user)
        await self.db.flush()
        await self.db.refresh(user)
        return user


class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def register(self, data: UserCreate) -> User:
        existing = await self.repo.find_by_email(data.email)
         existing:
             ValueError()
        user = User(
            email=data.email,
            display_name=data.display_name,
            hashed_password=hash_password(data.password),
        )
          .repo.save(user)

Pattern 2: Pagination Helper

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

T = TypeVar("T")


class PaginatedResponse(BaseModel, Generic[T]):
    data: Sequence[T]
    total: int
    page: int
    page_size: int
    has_next: bool
    has_prev: bool

    @classmethod
    def create(
        cls, data: Sequence[T], total: int, page: int, page_size: int
    ) -> "PaginatedResponse[T]":
        return cls(
            data=data,
            total=total,
            page=page,
            page_size=page_size,
            has_next=(page * page_size) < total,
            has_prev=page > 1,
        )

Common Rationalizations

Rationalization	Reality
"I will define the endpoint with a plain dict instead of a response_model; it is faster."	Without response_model, internal fields (password hashes, internal IDs) leak into the JSON response, and the OpenAPI schema no longer matches reality. The explicit model is what enforces the API contract and filters the output.
"It is just one synchronous DB call inside an async endpoint; it will not matter."	A blocking call inside an async def handler stalls the entire event loop, so one slow query freezes every concurrent request, not just its own. Use an async driver or offload to a thread pool.
"I will validate the request body with a few if checks instead of a Pydantic model."	Manual checks miss type coercion and edge cases Pydantic handles for free, and they do not appear in the generated docs. A malformed payload then reaches your business logic instead of being rejected at the boundary with a 422.

Verification

• All endpoints have explicit response_model definitions
• Input validation uses Pydantic schemas (not manual checks)
• Passwords are hashed (bcrypt/argon2), never stored in plain text
• JWT tokens have expiration times and are validated on every request
• Database sessions are managed via dependency injection with proper cleanup
• Background tasks do not block the event loop (use async or thread pool)
• CORS is configured with specific origins (not wildcard in production)
• Tests use dependency overrides for database isolation
• Health check endpoint verifies database connectivity
• Dockerfile uses non-root user and multi-stage build
• OpenAPI schema is reviewed and reflects actual API contract
• Rate limiting is applied to authentication endpoints

References

• references/dependency-injection-patterns.md - quick-lookup guide for FastAPI dependency injection patterns covering database sessions, authentication, pagination, and testing.

Related Skills

• [[python-cleanup]] - Python code quality and cleanup patterns
• [[unit-tests]] - General unit testing strategies
• [[api-documentation]] - API documentation standards
• [[kubernetes-expert]] - Container orchestration for API deployment
• [[security-review]] - Security review for API endpoints

---

Version: 1.0.0 Last Updated: March 2026

Iterative Refinement Strategy

This skill is optimized for an iterative approach:

1. Execute: Perform the core steps defined above.
2. Review: Critically analyze the output (coverage, quality, completeness).
3. Refine: If targets aren't met, repeat the specific implementation steps with improved context.
4. Loop: Continue until the definition of done is satisfied.

---

Source: bendourthe/Nexus-Hub — distributed by TomeVault.