Comprehensive Rules for architecting, coding, testing and deploying high-performance FastAPI back-ends with Pydantic v2.
Transform your API development from reactive debugging to proactive architecture with production-grade FastAPI patterns that eliminate common performance bottlenecks and maintenance headaches.
You're probably dealing with these familiar frustrations:
These aren't just minor inconveniences—they're productivity killers that turn every feature request into a potential system-wide debugging session.
These Cursor Rules establish a complete development framework that eliminates guesswork and prevents common FastAPI pitfalls before they reach production. Instead of learning through painful debugging sessions, you get battle-tested patterns that work at scale.
What you get:
# Typical FastAPI code that looks fine but fails under load
@app.get("/users/{user_id}")
def get_user(user_id: int):
# Blocking database call - freezes under concurrent requests
user = db.query(User).filter(User.id == user_id).first()
# No validation, returns ORM object directly
return user
# Clean, testable, performant FastAPI endpoint
@router.get("/{user_id}", response_model=UserOut, status_code=status.HTTP_200_OK)
async def get_user(
user_id: int,
svc: Annotated[UserService, Depends()]
):
return await svc.fetch_user(user_id)
Immediate improvements:
Before: Database changes require application restarts and careful coordination
# Your old approach - tightly coupled to ORM
def create_user(user_data: dict):
user = User(**user_data) # Hope the dict matches your model
db.add(user)
db.commit()
return user # Returns ORM object with circular references
After: Clean separation with async repositories
# Repository pattern with async operations
class UserRepository:
async def create_user(self, user_data: UserCreateDTO) -> UserDTO:
async with self.session() as session:
user_entity = User.from_dto(user_data)
session.add(user_entity)
await session.commit()
return UserDTO.from_entity(user_entity)
Before: Runtime errors from malformed requests
# Fragile endpoint that breaks in production
@app.post("/users")
def create_user(request: dict): # No validation
# Hope the client sent the right fields
email = request["email"] # KeyError waiting to happen
After: Fail-fast validation with detailed error responses
# Pydantic v2 with strict validation
class UserCreateRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
email: EmailStr
name: str = Field(min_length=1, max_length=100)
age: int = Field(ge=18, le=120)
@router.post("/", response_model=UserOut)
async def create_user(request: UserCreateRequest, svc: Annotated[UserService, Depends()]):
return await svc.create_user(request)
Before: Generic 500 errors that tell you nothing
# Unhelpful error handling
@app.get("/users/{user_id}")
def get_user(user_id: int):
try:
user = some_database_call(user_id)
return user
except Exception:
raise HTTPException(status_code=500, detail="Something went wrong")
After: Structured error responses with context
# Domain-specific error handling
@app.exception_handler(DomainError)
async def domain_error_handler(_, exc: DomainError):
logger.bind(error_code=exc.code, context=exc.context).error("Domain error occurred")
return JSONResponse(
status_code=400,
content={"error_code": exc.code, "detail": exc.message}
)
my_api/
├── app/
│ ├── users/
│ │ ├── __init__.py
│ │ ├── schemas.py # Pydantic models
│ │ ├── router.py # FastAPI routes
│ │ ├── service.py # Business logic
│ │ ├── repository.py # Data access
│ │ └── tests/ # Feature tests
│ ├── core/
│ │ ├── config.py # Environment configuration
│ │ ├── database.py # Async SQLAlchemy setup
│ │ └── exceptions.py # Custom exception classes
│ └── main.py # FastAPI app initialization
# requirements.txt focused on production readiness
fastapi[all]==0.104.1
pydantic[email]==2.5.0
sqlalchemy[asyncio]==2.0.23
alembic==1.12.1
asyncpg==0.29.0
uvicorn[standard]==0.24.0
pytest-asyncio==0.21.1
httpx==0.25.2
structlog==23.2.0
prometheus-fastapi-instrumentator==6.1.0
# main.py - Clean application initialization
from fastapi import FastAPI
from app.core.database import init_db
from app.users.router import router as users_router
app = FastAPI(title="Production API", version="1.0.0")
# Structured middleware stack
app.add_middleware(RequestTimeoutMiddleware, timeout=90)
app.add_middleware(CORSMiddleware, allow_origins=["*"])
app.add_middleware(GZIPMiddleware)
# Feature routers
app.include_router(users_router, prefix="/api/v1/users", tags=["users"])
@app.on_event("startup")
async def startup():
await init_db()
# conftest.py - Async testing setup
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.fixture
async def async_client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def db_session():
# Create test database session
async with async_sessionmaker() as session:
yield session
await session.rollback()
# Dockerfile - Multi-stage production build
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
FROM python:3.11-slim
RUN useradd --create-home --shell /bin/bash app
USER app
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
Stop fighting FastAPI's async nature and start leveraging it. These rules transform your development process from reactive debugging to proactive architecture, giving you the confidence to ship features fast without breaking production.
Your next API deployment will be your smoothest yet.
You are an expert in Python 3.11+, FastAPI, Pydantic v2, AsyncIO, SQLAlchemy 2.0 (async), Alembic, PyTest, httpx, Docker, Kubernetes and GitHub Actions.
# Key Principles
- Async-first and I/O-bound design: every blocking call must be replaced with an async-aware alternative.
- Clean, layered architecture (routers → services → repositories → external systems).
- Single-responsibility modules; never mix I/O, business logic and presentation in the same layer.
- Strong static typing everywhere; `mypy` must pass with `--strict`.
- Favor composition and pure functions over inheritance and global state.
- Immutable input, explicit output: never mutate Pydantic request models.
- Fail fast: validate early, return early, keep the “happy path” last in a function.
- Observability built-in: structured JSON logging and Prometheus metrics from day 1.
# Python Specific Rules
- Minimum version: 3.11. Use `taskgroups`, `typing.Self`, `typeGuard`, `match`.
- Use `ruff` with the following profiles enabled: `flake8`, `isort`, `black` (line length 88), `pydantic`, `asyncio`.
- File layout per feature:
`my_app/<feature>/ __init__.py, schemas.py, router.py, service.py, repository.py, tests/`.
- Naming:
• modules & packages: `snake_case`
• classes: `PascalCase`
• functions/vars: `snake_case`
• async functions MUST be prefixed with a verb: `fetch_`, `create_`, `update_`.
- Use `Annotated` for dependency injection metadata: `user: Annotated[User, Depends(get_current_user)]`.
- Use Pydantic v2 `BaseModel` for external contracts, `@dataclass(slots=True, frozen=True)` for internal DTOs.
- Never store ORM entities in Pydantic models; instead convert in/out with `model_validate` / `model_dump`.
- Default to `Enum`, `Literal` and `typing.TypedDict` to avoid magic strings.
# Error Handling & Validation
- Validate all incoming payloads with Pydantic; forbid extra fields via `model_config = ConfigDict(extra="forbid")`.
- Centralise exception mapping:
```python
@app.exception_handler(DomainError)
async def domain_error_handler(_, exc: DomainError):
return JSONResponse(status_code=400, content={"error_code": exc.code, "detail": exc.message})
```
- Never raise bare `HTTPException` outside router layer; convert domain errors in service layer.
- Use `HTTP_4xx` for client mistakes, `HTTP_5xx` exclusively for unhandled errors.
- Log errors with context:
`logger.bind(request_id=req_id, path=request.url.path).exception("Unhandled")`.
- Implement health check endpoints (`/health/live`, `/health/ready`) returning 200/503.
# FastAPI Rules
- Instantiate app in `main.py`; import nothing from `main.py` elsewhere to avoid circular deps.
- Routers
• Each feature exposes an `APIRouter` object.
• Use `tags=["users"]`, `summary=`, `response_model=` for auto-docs.
• Mount at `/api/v1/<feature>`.
- Dependencies
• Use `Depends(get_db)` pattern for session injection.
• Re-use your service classes inside routers; never touch DB in routers.
- Background work
• Use `BackgroundTasks` for small, I/O tasks (<1 s).
• CPU-bound tasks go to workers via Celery/RQ.
- Security
• OAuth2 Password/Bearer with JWT (RS256).
• Keep token TTL short (15 min) + refresh token cookie.
• Validate scope/roles inside dependency, not in route.
- Middleware
• Add request-id, timeout (90 s), CORS configured from env list, and GZip.
# SQLAlchemy 2.0 Async Rules
- Use `async_engine = create_async_engine(DB_URL, pool_pre_ping=True)`.
- Session dependency:
```python
async def get_session() -> AsyncIterator[AsyncSession]:
async_sessionmaker = sessionmaker(async_engine, expire_on_commit=False, class_=AsyncSession)
async with async_sessionmaker() as session:
yield session
```
- Repositories return domain DTOs, not ORM entities.
- Migrations via Alembic’s async support (`script.py.mako` update env.py with async engine`).
# Testing
- Use `pytest-asyncio` + `httpx.AsyncClient`.
- Directory `tests/feature/test_<endpoint>.py` per router.
- Fixtures: `event_loop`, `async_client`, `db_session`, `fake_user`.
- 100 % schema coverage: each endpoint has success + failure tests.
- Contract testing: generate OpenAPI, diff against previous in CI.
# Performance Optimization
- Keep every DB call under 50 ms; use `sqlalchemy.orm.selectinload`.
- Cache hot GETs with `aiocache` + Redis, TTL = 60 s.
- Use `uvicorn main:app --workers $(nproc) --loop uvloop --http h11` in prod.
- Timeouts: client-read 15 s, client-write 15 s, keep-alive 75 s.
# Security
- Run `bandit`, `safety`, `pip-audit` in CI.
- Enable `SECURE_*` headers via Starlette’s `SecurityMiddleware`.
- Enforce HTTPS (HSTS = 31536000, preload).
- Use Let’s Encrypt certs auto-renew via cert-manager on K8s.
- Rotate JWT signing keys every 30 days; keep 2 keys for overlap.
# Deployment & CI/CD
- Dockerfile guidelines:
• `python:3.11-slim`, multi-stage, install build deps then `pip install --no-cache-dir .[prod]`.
• `USER app:app`, `WORKDIR /app`.
• Expose port 8000, CMD use `uvicorn`.
- K8s:
• Liveness `/health/live`, readiness `/health/ready`.
• Resources: requests 100m/128Mi, limits 500m/512Mi.
• Use `HPA` on `cpu+requests`.
- GitHub Actions pipeline:
1. `ruff`, `mypy --strict`, `pytest -q`.
2. `docker buildx build --push` to GHCR.
3. Trivy image scan must pass.
4. ArgoCD sync to cluster.
# Logging & Monitoring
- Use `structlog` with JSON renderer; include timestamp, level, event, request_id, user_id.
- Forward logs to Loki; dashboards in Grafana.
- Use Prometheus FastAPI instrumentation: latency, throughput, in-flight.
- Alert on p95 > 300 ms, 5xx rate > 1 %.
# Common Pitfalls
- Forgetting `await` on DB ops ⇒ app freezes. Always run `pytest --asyncio-mode=strict` to catch.
- Returning ORM instances in response ⇒ circular reference & un-awaitable lazy loads. Always map to schema.
- Running blocking CPU work (e.g. bcrypt) in route ⇒ use `run_in_threadpool` or Celery.
- Not closing DB sessions ⇒ connection leak. Use dependency shown above.
# Linting & Formatting
- Run `pre-commit` with hooks: `black`, `ruff`, `mypy`, `pyproject-sort`, `detect-secrets`.
# Documentation
- Keep `/docs` enabled only in non-prod; deploy Redoc static docs to S3 for prod.
- Tag endpoints meaningfully; supply `operation_id` for each route.
# Versioning
- Prefix all stable routes with `/api/v{n}`. Bump major only on breaking change.
# Ready-to-Use Example Snippet
```python
from fastapi import APIRouter, Depends, status
from my_app.user.schemas import UserOut
from my_app.user.service import UserService
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/{user_id}", response_model=UserOut, status_code=status.HTTP_200_OK)
async def get_user(user_id: int, svc: UserService = Depends()):
return await svc.fetch_user(user_id)
```