This document provides guidelines for AI agents working on FastAPI projects. Follow these conventions when writing or modifying code.
Organize code by domain, not by file type.
src/
├── {domain}/ # e.g., auth/, posts/, aws/
│ ├── router.py # API endpoints
│ ├── schemas.py # Pydantic models
│ ├── models.py # Database models
│ ├── service.py # Business logic
│ ├── dependencies.py # Route dependencies
│ ├── config.py # Environment variables
│ ├── constants.py # Constants and error codes
│ ├── exceptions.py # Domain-specific exceptions
│ └── utils.py # Helper functions
├── config.py # Global configuration
├── models.py # Global models
├── exceptions.py # Global exceptions
├── database.py # Database connection
└── main.py # FastAPI app initialization
Import Convention: Use explicit module names when importing across domains:
from src.auth import constants as auth_constants
from src.notifications import service as notification_serviceasync defroutes: Use ONLY non-blocking I/O (awaitcalls)defroutes (sync): Use for blocking I/O (runs in threadpool automatically)- CPU-intensive work: Offload to Celery or multiprocessing
# WRONG: Blocking call in async route
@router.get("/bad")
async def bad_route():
time.sleep(10) # Blocks entire event loop
return {"status": "done"}
# CORRECT: Non-blocking in async route
@router.get("/good")
async def good_route():
await asyncio.sleep(10)
return {"status": "done"}
# CORRECT: Sync route for blocking operations
@router.get("/also-good")
def sync_route():
time.sleep(10) # Runs in threadpool
return {"status": "done"}from fastapi.concurrency import run_in_threadpool
@router.get("/")
async def call_sync_library():
result = await run_in_threadpool(sync_client.make_request, data=my_data)
return resultfrom pydantic import BaseModel, EmailStr, Field
class UserCreate(BaseModel):
username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$")
email: EmailStr
age: int = Field(ge=18)Create a shared base model for consistent serialization:
from pydantic import BaseModel, ConfigDict
class CustomModel(BaseModel):
model_config = ConfigDict(
json_encoders={datetime: datetime_to_gmt_str},
populate_by_name=True,
)# src/auth/config.py
class AuthConfig(BaseSettings):
JWT_ALG: str
JWT_SECRET: str
JWT_EXP: int = 5
auth_settings = AuthConfig()async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
post = await service.get_by_id(post_id)
if not post:
raise PostNotFound()
return post
@router.get("/posts/{post_id}")
async def get_post(post: dict[str, Any] = Depends(valid_post_id)):
return postasync def valid_owned_post(
post: dict[str, Any] = Depends(valid_post_id),
token_data: dict[str, Any] = Depends(parse_jwt_data),
) -> dict[str, Any]:
if post["creator_id"] != token_data["user_id"]:
raise UserNotOwner()
return post- Dependencies are cached per request (same dependency called multiple times = one execution)
- Prefer
asyncdependencies to avoid threadpool overhead - Use consistent path variable names to enable dependency reuse
Use consistent path variable names for dependency reuse:
# Both use profile_id, enabling shared valid_profile_id dependency
GET /profiles/{profile_id}
GET /creators/{profile_id}- Use
lower_case_snakeformat - Singular table names:
post,user,post_like - Group related tables with prefix:
payment_account,payment_bill - DateTime suffix:
_at(e.g.,created_at) - Date suffix:
_date(e.g.,birth_date)
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",
}Prefer database-level operations for:
- Complex joins
- Data aggregation
- Building nested JSON responses
- Keep migrations static and reversible
- Use descriptive file names:
2022-08-24_post_content_idx.py - Configure in alembic.ini:
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s
SHOW_DOCS_ENVIRONMENT = ("local", "staging")
app_configs = {"title": "My API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
app_configs["openapi_url"] = None
app = FastAPI(**app_configs)@router.post(
"/endpoints",
response_model=DefaultResponseModel,
status_code=status.HTTP_201_CREATED,
description="Description of the endpoint",
tags=["Category"],
responses={
status.HTTP_201_CREATED: {"model": CreatedResponse},
status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse},
},
)Use async test client from the start:
import pytest
from httpx import AsyncClient, ASGITransport
@pytest.fixture
async def client():
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
yield client
@pytest.mark.asyncio
async def test_endpoint(client: AsyncClient):
resp = await client.post("/posts")
assert resp.status_code == 201Use ruff for formatting and linting:
ruff check --fix src
ruff format src- Apply SOLID principles when writing Python code relentlessly.
- When considering patterns that adhere to those principles, prefer leveraging Gang of Four patterns over less canonical design patterns
- Apply DRY relentlessly
- Be sure all magic numbers are constantized
- Ensure the schema is in 3NF
If there is a more canonical to way express, or describe, a facet of the domain, wherever it exists, or will exist, throughout the repo, prefer using that language to describe it as opposed to a less canonical term.
| Scenario | Solution |
|---|---|
| Non-blocking I/O | async def route with await |
| Blocking I/O | def route (sync) |
| Sync library in async | run_in_threadpool() |
| CPU-intensive | Celery/multiprocessing |
| Request validation | Dependencies with DB checks |
| Shared validation | Chain dependencies |
| Config per domain | Separate BaseSettings classes |
| Complex DB queries | SQL with JSON aggregation |