Claude fastapi skill: 10 backend endpoints you can ship

The prompt

Scaffold a `users` resource using the fastapi-templates skill's layered layout (api/, core/, models/, schemas/, services/, repositories/). Generate `repositories/base_repository.py` as a generic async `BaseRepository[ModelType, CreateSchemaType, UpdateSchemaType]` with `get`, `get_multi(skip, limit)`, `create`, `update(exclude_unset)`, and `delete`. Use SQLAlchemy 2.0 `select()` + `await db.execute(...)`, never the legacy `db.query()`. Then a `UserRepository(BaseRepository)` subclass with `get_by_email`. Wire the endpoints in `api/v1/endpoints/users.py` through `Depends(get_db)`.

What slides.md looks like

# repositories/base_repository.py
from typing import Generic, TypeVar, Type, Optional, List
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel

ModelType = TypeVar("ModelType")
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)

class BaseRepository(Generic[ModelType, CreateSchemaType]):
    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
        result = await db.execute(select(self.model).where(self.model.id == id))
        return result.scalars().first()

    async def get_multi(self, db: AsyncSession, skip: int = 0, limit: int = 100):
        result = await db.execute(select(self.model).offset(skip).limit(limit))
        return result.scalars().all()

Pairs with

The prompt

Add OAuth2 password-flow auth following the fastapi-templates security pattern. In `core/security.py`: `pwd_context = CryptContext(schemes=['bcrypt'])`, `create_access_token(data, expires_delta)` signing with `settings.SECRET_KEY` and HS256, plus `verify_password` and `get_password_hash`. In `api/dependencies.py`: `oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/v1/auth/login')` and a `get_current_user` dependency that decodes the token, reads `sub`, and loads the user from the repository — raising a 401 with a `WWW-Authenticate: Bearer` header on any JWTError.

What slides.md looks like

# core/security.py
from datetime import datetime, timedelta
from jose import jwt
from passlib.context import CryptContext
from app.core.config import get_settings

settings = get_settings()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

Pairs with

The prompt

Generate `schemas/user.py` with Pydantic v2 models: `UserBase` (email: EmailStr, name: str), `UserCreate(UserBase)` adding `password: str` with `Field(min_length=8)`, `UserUpdate` with every field `Optional` for PATCH, and a read `User(UserBase)` that adds `id: int` and `is_active: bool` and sets `model_config = ConfigDict(from_attributes=True)`. The read model must never include the password. Use `field_validator` to lowercase the email on input.

What slides.md looks like

# schemas/user.py
from pydantic import BaseModel, EmailStr, Field, ConfigDict, field_validator

class UserBase(BaseModel):
    email: EmailStr
    name: str

    @field_validator("email")
    @classmethod
    def lower_email(cls, v: str) -> str:
        return v.lower()

class UserCreate(UserBase):
    password: str = Field(min_length=8)

class User(UserBase):
    id: int
    is_active: bool
    model_config = ConfigDict(from_attributes=True)

Pairs with

The prompt

Generate `core/database.py` per the fastapi-templates pattern: `create_async_engine(settings.DATABASE_URL, future=True)`, an `AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)`, a `Base = declarative_base()`, and a `get_db()` async generator dependency that yields a session, commits on success, rolls back inside an `except` block, and closes in `finally`. Show how an endpoint consumes it with `db: AsyncSession = Depends(get_db)`.

What slides.md looks like

# core/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.core.config import get_settings

settings = get_settings()
engine = create_async_engine(settings.DATABASE_URL, echo=True, future=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Pairs with

The prompt

Add a `POST /api/v1/users/` flow that creates the user, then schedules a welcome email via FastAPI `BackgroundTasks` so the request returns 201 without waiting on SMTP. Show the `background_tasks: BackgroundTasks` parameter and `background_tasks.add_task(send_welcome_email, user.email)`. Make `send_welcome_email` an async function. Add a one-line note on when this is the wrong tool — i.e. when the work must survive a process restart, reach for a real queue.

What slides.md looks like

# api/v1/endpoints/users.py
from fastapi import APIRouter, BackgroundTasks, Depends, status
from app.services.email_service import send_welcome_email

router = APIRouter()

@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
):
    user = await user_service.create_user(db, user_in)
    background_tasks.add_task(send_welcome_email, user.email)
    return user

Pairs with

The prompt

Register global exception handlers on the app. Add `@app.exception_handler(ValueError)` returning a 400 JSONResponse with `{'detail': str(exc)}`, override the default `RequestValidationError` handler to log and return 422, and a catch-all `Exception` handler that logs the traceback and returns a generic 500 without leaking internals. Keep the service layer raising plain `ValueError` / domain exceptions so endpoints don't each wrap their own try/except.

What slides.md looks like

# core/handlers.py
import logging
from fastapi import Request, status
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

def register_handlers(app):
    @app.exception_handler(ValueError)
    async def value_error_handler(request: Request, exc: ValueError):
        return JSONResponse(status_code=400, content={"detail": str(exc)})

    @app.exception_handler(Exception)
    async def unhandled_handler(request: Request, exc: Exception):
        logger.exception("Unhandled error on %s", request.url.path)
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"detail": "Internal server error"},
        )

Pairs with

The prompt

Generate the fastapi-templates testing setup: a `conftest.py` with an async `db_session` fixture using `sqlite+aiosqlite:///:memory:`, a `client` fixture that overrides `app.dependency_overrides[get_db]` to yield the test session and wraps the app in `httpx.AsyncClient(app=app, base_url='http://test')`, and a `test_users.py` that `await client.post('/api/v1/users/', json={...})` and asserts the 201 response shape. Mark async tests with `@pytest.mark.asyncio`.

What slides.md looks like

# tests/conftest.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import get_db, Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(app=app, base_url="http://test") as c:
        yield c

Pairs with

The prompt

Add pagination and filtering to `GET /api/v1/items/`. Accept `skip: int = Query(0, ge=0)`, `limit: int = Query(20, ge=1, le=100)`, and optional `status: str | None = Query(None)`. Extend the repository with a `get_filtered(db, skip, limit, status)` that conditionally adds `.where(Item.status == status)` to the `select()`. Return a `{'items': [...], 'total': int, 'skip': skip, 'limit': limit}` envelope so the client can render pagination.

What slides.md looks like

# api/v1/endpoints/items.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select, func

@router.get("/")
async def list_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    db: AsyncSession = Depends(get_db),
):
    stmt = select(Item)
    if status:
        stmt = stmt.where(Item.status == status)
    total = await db.scalar(select(func.count()).select_from(stmt.subquery()))
    rows = (await db.execute(stmt.offset(skip).limit(limit))).scalars().all()
    return {"items": rows, "total": total, "skip": skip, "limit": limit}

Pairs with

The prompt

Write a production `Dockerfile` for the fastapi-templates project. Use a `python:3.12-slim` base, install dependencies from `requirements.txt` in a cached layer before copying source, run as a non-root user, expose 8000, and start with `gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 --bind 0.0.0.0:8000`. Add a `.dockerignore` and a `HEALTHCHECK` that curls `/health`. Explain in one line why gunicorn-with-uvicorn-workers beats a single uvicorn process for production.

What slides.md looks like

# Dockerfile
FROM python:3.12-slim
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN useradd -m appuser && chown -R appuser /app
USER appuser

EXPOSE 8000
HEALTHCHECK --interval=30s CMD curl -f http://localhost:8000/health || exit 1
CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker", \
     "-w", "4", "--bind", "0.0.0.0:8000"]

Pairs with

The prompt

Add API-key middleware to the FastAPI app. Register a `@app.middleware('http')` function that reads `X-API-Key`, compares it against `settings.API_KEYS` (a set), and returns a 401 JSONResponse for protected paths — but skips the check for `/docs`, `/redoc`, `/openapi.json`, and `/health` so the interactive docs stay open. Keep FastAPI's auto-generated OpenAPI schema intact, and document the header with an `APIKeyHeader` security scheme so it shows up in /docs.

What slides.md looks like

# core/middleware.py
from fastapi import Request, status
from fastapi.responses import JSONResponse

OPEN_PATHS = {"/docs", "/redoc", "/openapi.json", "/health"}

def add_api_key_middleware(app, settings):
    @app.middleware("http")
    async def api_key_guard(request: Request, call_next):
        if request.url.path in OPEN_PATHS:
            return await call_next(request)
        if request.headers.get("X-API-Key") not in settings.API_KEYS:
            return JSONResponse(status_code=401, content={"detail": "Invalid API key"})
        return await call_next(request)

Pairs with