xpro-test

Imported from SN4S/xpro-test (AGENTS.md). Install upstream with npx skills add SN4S/xpro-test. Copyright stays with the author.

FastAPI Best Practices for AI Agents

A machine-readable companion to README.md for AI coding agents working in FastAPI projects. Same rules, restructured for fast pattern matching: version pins, Do/Don't blocks, anti-patterns, and a quick-reference table.

Compatibility Matrix

Pin to these versions or newer. Examples in this file assume them.

Project Structure

Organize by domain, not by file type. One package per bounded context.

src/
├── {domain}/           # e.g., auth/, posts/, aws/
│   ├── router.py       # API endpoints
│   ├── schemas.py      # Pydantic models
│   ├── models.py       # SQLAlchemy ORM models
│   ├── service.py      # Business logic
│   ├── dependencies.py # Route dependencies
│   ├── config.py       # Domain-scoped BaseSettings
│   ├── constants.py    # Constants and error codes
│   ├── exceptions.py   # Domain-specific exceptions
│   └── utils.py        # Helper functions
├── config.py           # Global BaseSettings
├── models.py           # Shared Pydantic / ORM bases
├── exceptions.py       # Global exceptions
├── database.py         # Async engine + session factory
└── main.py             # FastAPI app + lifespan

Cross-domain imports: always use the explicit module name. Never from src.auth import *.

from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode

Async Routes

Decision rule

Do / Don't

# DON'T — blocking call inside async route freezes the entire event loop
@router.get("/bad")
async def bad():
    time.sleep(10)            # blocks every request on this worker
    return {"ok": True}

# DO — sync route lets FastAPI run it in a threadpool
@router.get("/sync-ok")
def sync_ok():
    time.sleep(10)            # blocks one threadpool worker, not the loop
    return {"ok": True}

# DO — async route with awaitable sleep
@router.get("/async-ok")
async def async_ok():
    await asyncio.sleep(10)   # yields control, loop keeps serving requests
    return {"ok": True}

# DO — async route that has to call a sync library
from fastapi.concurrency import run_in_threadpool

@router.get("/wrap")
async def wrap():
    result = await run_in_threadpool(legacy_sync_client.fetch, "id")
    return result

Threadpool caveats

• Default Starlette threadpool size is 40. Saturating it slows every sync route.
• Threads cost more than coroutines. Don't use sync routes "just because."

Pydantic

Use built-in validators

from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field


class MusicBand(StrEnum):
    AEROSMITH = "AEROSMITH"
    QUEEN = "QUEEN"
    ACDC = "AC/DC"


class UserCreate(BaseModel):
    first_name: str = Field(min_length=1, max_length=128)
    username: str = Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_-]+$")
    email: EmailStr
    age: int = Field(ge=18)                     # required, must be >= 18
    favorite_band: MusicBand | None = None
    website: AnyUrl | None = None

Don't write Field(ge=18, default=None). The constraint and the default contradict each other. Decide: required (Field(ge=18)) or optional (int | None = Field(default=None, ge=18)).

Custom base model — modern serialization

json_encoders is deprecated in Pydantic v2. Use @field_serializer for per-field rules, or annotate a custom type with PlainSerializer.

from datetime import datetime
from zoneinfo import ZoneInfo
from pydantic import BaseModel, ConfigDict, field_serializer


class CustomModel(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    @field_serializer("*", when_used="json", check_fields=False)
    def _serialize_datetimes(self, value):
        if isinstance(value, datetime):
            if value.tzinfo is None:
                value = value.replace(tzinfo=ZoneInfo("UTC"))
            return value.strftime("%Y-%m-%dT%H:%M:%S%z")
        return value

Split BaseSettings by domain

pydantic-settings is its own package since Pydantic v2.

# src/auth/config.py
from datetime import timedelta
from pydantic_settings import BaseSettings, SettingsConfigDict


class AuthConfig(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="AUTH_", env_file=".env", extra="ignore")

    JWT_ALG: str
    JWT_SECRET: str
    JWT_EXP_MINUTES: int = 5
    REFRESH_TOKEN_KEY: str
    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
    SECURE_COOKIES: bool = True


auth_settings = AuthConfig()

Dependencies

Use Annotated, not default-arg Depends(...)

Annotated[T, Depends(...)] is the idiomatic form since FastAPI 0.95 and avoids gotchas with default values.

# DO — modern Annotated form
from typing import Annotated
from fastapi import Depends

PostDep = Annotated[dict, Depends(valid_post_id)]

@router.get("/posts/{post_id}")
async def get_post(post: PostDep):
    return post

# Avoid — default-argument form (still works, but legacy)
@router.get("/posts/{post_id}")
async def get_post(post: dict = Depends(valid_post_id)):
    return post

Validate inside dependencies (not just inject)

async def valid_post_id(post_id: UUID4) -> dict:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()
    return post

Chain dependencies for reuse

async def valid_owned_post(
    post: Annotated[dict, Depends(valid_post_id)],
    token_data: Annotated[dict, Depends(parse_jwt_data)],
) -> dict:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()
    return post

Rules

• Dependencies are cached per request. Same Depends(x) called 5 times in one request → x runs once.
• Prefer async def dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks.
• Use the same path-variable name across endpoints when you want to share a dependency (e.g. profile_id in both /profiles/{profile_id} and /creators/{profile_id}).

Authentication — JWT

Use PyJWT, not python-jose (unmaintained).

import jwt  # PyJWT
from jwt.exceptions import InvalidTokenError

def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])
    except InvalidTokenError as exc:
        raise InvalidCredentials() from exc

Database — SQLAlchemy 2.0 async

Prefer SQLAlchemy 2.0's async API. encode/databases is in maintenance mode — don't pick it for new projects.

# src/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


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

Naming conventions

• lower_case_snake
• Singular tables: post, user, post_like
• Group with prefix: payment_account, payment_bill
• _at suffix for datetime, _date suffix for date
• Use the same FK column name everywhere it appears (profile_id, not user_id in some tables and profile_id in others)

Index naming convention

from sqlalchemy import MetaData

POSTGRES_INDEXES_NAMING_CONVENTION = {
    "ix": "%(column_0_label)s_idx",
    "uq": "%(table_name)s_%(column_0_name)s_key",
    "ck": "%(table_name)s_%(constraint_name)s_check",
    "fk": "%(table_name)s_%(column_0_name)s_fkey",
    "pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)

SQL-first, Pydantic-second

• Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this.
• Hydrate the result into Pydantic only for response validation, not for transformation.

Background work — BackgroundTasks vs Celery

from fastapi import BackgroundTasks

@router.post("/signup")
async def signup(data: SignupIn, bg: BackgroundTasks):
    user = await service.create_user(data)
    bg.add_task(send_welcome_email, user.email)   # fire-and-forget, in-process
    return user

BackgroundTasks run after the response is sent, in the same worker process. If the worker dies, the task is lost. There is no retry. Don't use them for anything you'd page on.

Testing

Async client from day one

import pytest
from httpx import AsyncClient, ASGITransport

from src.main import app


@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_create_post(client: AsyncClient):
    resp = await client.post("/posts", json={"title": "hi"})
    assert resp.status_code == 201

Don't use async_asgi_testclient — it's unmaintained. The example above (httpx + ASGITransport) is the supported path.

Override dependencies in tests

Don't monkeypatch internals. Use FastAPI's built-in dependency_overrides.

from src.auth.dependencies import parse_jwt_data
from src.main import app


def fake_user():
    return {"user_id": "00000000-0000-0000-0000-000000000001"}


@pytest.fixture(autouse=True)
def _override_auth():
    app.dependency_overrides[parse_jwt_data] = fake_user
    yield
    app.dependency_overrides.clear()

Migrations (Alembic)

• Migrations must be static and reversible.
• Use the async template: alembic init -t async migrations
• Descriptive filenames:

  # alembic.ini
  file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s

  → 2026-04-14_add_post_content_idx.py

API documentation

Hide docs outside selected envs

from fastapi import FastAPI
from src.config import settings

SHOW_DOCS_IN = {"local", "staging"}
app_kwargs = {"title": "My API"}
if settings.ENVIRONMENT not in SHOW_DOCS_IN:
    app_kwargs["openapi_url"] = None    # disables /docs and /redoc

app = FastAPI(**app_kwargs)

Document endpoints fully

from fastapi import APIRouter, status

router = APIRouter()


@router.post(
    "/items",
    response_model=ItemResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create an item",
    description="Creates an item owned by the authenticated user.",
    tags=["items"],
    responses={
        status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse, "description": "Validation error"},
        status.HTTP_409_CONFLICT:    {"model": ErrorResponse, "description": "Slug already exists"},
    },
)
async def create_item(payload: ItemCreate) -> ItemResponse: ...

Linting

ruff check --fix src
ruff format src

Add to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.

Anti-patterns — common AI-agent mistakes

If you're an agent reviewing a diff, check for these. Each is a real failure mode I've seen agents introduce.

Quick reference