Cursor Rules for FastAPI: Production Rules for AI-Assisted Python API Development

Cursor Rules for FastAPI: The Complete Guide to AI-Assisted Python API Development

FastAPI is the framework where "the response came back with a 200" hides the longest lie. The endpoint returns, the swagger docs render, and nothing in uvicorn --reload tells you that the Pydantic model accepted an extra role: "admin" field the client hand-crafted and silently promoted the caller, that the Depends(get_db) handed out the same SQLAlchemy session to two concurrent requests because the yielder closed over a module-level engine, or that the "async" handler calls a synchronous requests.get that pegs the event loop while sixty requests queue behind it. The service ships. A week later a pager fires because every /health check is now 8 seconds.

Then you add an AI assistant.

Cursor and Claude Code were trained on a planet's worth of Python, most of it Flask, pre-Pydantic-v2, pre-type-hints-as-API, and most of the FastAPI in training data is the tutorial "to-do list" app. Ask for "an endpoint that lists users with pagination and filtering," and you get a handler with dict return types, an inline SQLAlchemy session created with SessionLocal() inside the function, a try/except Exception that returns {"error": str(e)} with a 200, and a pagination scheme that offsets by row count on a million-row table. It runs. It's not the FastAPI you would ship in 2026.

The fix is .cursorrules — one file in the repo that tells the AI what idiomatic modern FastAPI looks like. Eight rules below, each with the failure mode, the rule, and a before/after. Copy-paste .cursorrules at the end.

---

How Cursor Rules Work for FastAPI Projects

Cursor reads project rules from two locations: .cursorrules (a single file at the repo root, still supported) and .cursor/rules/*.mdc (modular files with frontmatter, recommended for anything larger than a single main.py). For FastAPI I recommend modular rules so an internal admin API's auth conventions don't bleed into a public API's rate-limit rules:

.cursor/rules/
  fastapi-core.mdc        # Pydantic v2, response_model, status codes
  fastapi-db.mdc          # async SQLAlchemy, session-per-request
  fastapi-auth.mdc        # OAuth2, JWT, scopes, dependencies
  fastapi-errors.mdc      # exception handlers, problem-details
  fastapi-testing.mdc     # pytest-asyncio, httpx.AsyncClient, factories

Frontmatter controls activation: globs: ["**/*.py"] with alwaysApply: false. Now the rules.

---

Rule 1: Pydantic v2 Schemas — Strict, Versioned, Separate From ORM Models

The most common AI failure in FastAPI is "the ORM model is the API model." Cursor returns a SQLAlchemy User from a handler and calls it done — now every column (password_hash, internal_notes, deleted_at) is in the JSON response, every refactor of the table is a public-API breaking change, and the extra="allow" default on older Pydantic lets the client hand-post a role="admin" field that **data splats into the ORM constructor. Pydantic v2 with model_config = ConfigDict(extra="forbid", strict=True) and separate request/response/DB schemas is the baseline.

The rule:

Three schema kinds per resource, all inheriting from a common base:
  - `UserBase` with shared fields
  - `UserCreate` / `UserUpdate` (request bodies) — no id, no server-owned fields
  - `UserRead` (response) — what is safe to return to clients
  - Internal `UserInDB` or a separate ORM class — never returned directly

All schemas configure:
  model_config = ConfigDict(
      extra="forbid",              # reject unknown fields
      strict=True,                 # no int->str coercion
      str_strip_whitespace=True,
      from_attributes=True,        # only where reading from ORM
  )

Every endpoint declares `response_model=...` explicitly. The return-type
annotation is a Pydantic model, not `dict` or `Any`. Never a bare ORM
instance.

Pagination is a typed schema — `Page[UserRead]` with `items`, `total`,
`page`, `size` — generated via `fastapi-pagination` or a hand-rolled
`Generic[T]`. Never `list[dict]` with an integer count alongside.

Field validation uses `Annotated[..., Field(..., min_length=3)]` and
typed validators (`@field_validator`). No runtime `if not x: raise`
scattered in handlers for shape validation.

Before — ORM leaks, untyped dict return, accepts extra fields:

@app.post("/users")
def create_user(data: dict, db: Session = Depends(get_db)):
    u = User(**data)
    db.add(u); db.commit(); db.refresh(u)
    return u  # returns every column including password_hash

After — strict schemas, explicit response_model, typed return:

class UserBase(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True, str_strip_whitespace=True)
    email: EmailStr
    name: Annotated[str, Field(min_length=1, max_length=80)]

class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=12, max_length=128)]

class UserRead(UserBase):
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    created_at: datetime

@router.post("/users", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
    body: UserCreate,
    db: AsyncSession = Depends(get_db),
) -> UserRead:
    user = await users_repo.create(db, body)
    return UserRead.model_validate(user)

Request with {"role": "admin"} now returns 422. password_hash is never in the response.

---

Rule 2: Async Everything — Or Explicitly Sync. Never Mix the Default Wrong

FastAPI runs async def handlers on the event loop and def handlers in a worker threadpool. Cursor's default is to mix them randomly, and when a def handler calls await db.execute(...) on an AsyncSession, or an async def handler calls a blocking requests.get(...), you get either a coroutine-never-awaited warning (at best) or a blocked event loop serving 1 request at a time (at worst). The rule is: be consistent per resource, pick async for new I/O-bound code, and never call blocking libraries from async def.

The rule:

New routers are `async def` throughout. Legacy sync routers are explicit
`def` — never mixed.

Blocking libraries (requests, psycopg2, sync redis, synchronous boto3)
are banned from async handlers. Use httpx.AsyncClient, asyncpg /
SQLAlchemy 2.0 AsyncSession, redis.asyncio, aioboto3.

If a blocking call is unavoidable (legacy SDK), wrap with
`await asyncio.to_thread(fn, *args)` — not `loop.run_in_executor`,
not `starlette.concurrency.run_in_threadpool` directly (ok but
`to_thread` is the stdlib-preferred spelling in 3.9+).

HTTP client lifecycle: one `httpx.AsyncClient` per app, kept alive in
`lifespan` context manager. Never `httpx.AsyncClient()` per request.

`time.sleep` is forbidden anywhere in request path. Use `await asyncio.sleep`.

`asyncio.gather` for fan-out with `return_exceptions=True` + explicit
error collection — not `asyncio.wait` without `return_when`.

CPU-bound work goes to a process pool (ProcessPoolExecutor) or a
background worker (Celery, Arq, Dramatiq) — never `await` an expensive
synchronous function on the event loop.

Before — async handler calling blocking library, per-request client creation:

@app.get("/weather/{city}")
async def weather(city: str):
    import requests
    resp = requests.get(f"https://api.weather.example/{city}", timeout=5)
    return resp.json()

Blocks the event loop for the duration of the HTTP call. Every request creates a fresh TCP pool.

After — lifespan-managed async client, non-blocking call:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(5.0, connect=2.0),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    )
    try:
        yield
    finally:
        await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

@router.get("/weather/{city}", response_model=WeatherRead)
async def weather(city: str, request: Request) -> WeatherRead:
    client: httpx.AsyncClient = request.app.state.http
    resp = await client.get(f"https://api.weather.example/{city}")
    resp.raise_for_status()
    return WeatherRead.model_validate(resp.json())

Event loop is not blocked. Connection pool is reused across requests.

---

Rule 3: Database Sessions Are Request-Scoped Dependencies — Not Module Globals

Cursor's default database pattern is SessionLocal = sessionmaker(engine); db = SessionLocal() at module level, used directly in handlers. Under concurrency this yields race conditions, session leaks, and DetachedInstanceError surfacing intermittently. The only correct pattern in FastAPI is a dependency that yields a session, auto-closes on response, and — for async — uses AsyncSession with expire_on_commit=False.

The rule:

Database access goes through a `get_db` dependency that yields an
`AsyncSession` (or sync `Session` in legacy codebases) and closes it
after the response is sent.

engine = create_async_engine(settings.DATABASE_URL, pool_pre_ping=True, pool_size=20)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

Handlers take `db: AsyncSession = Depends(get_db)` — never a module-level
session.

Transaction boundaries are explicit. Either:
  - Use the session's implicit transaction and commit at handler end
  - Wrap mutation blocks in `async with db.begin():` and let SQLAlchemy
    commit/rollback
Never partial commits inside a loop without a savepoint.

Queries live in a repository module (`users_repo.get_by_email(db, email)`)
— handlers orchestrate, they do not build SQL. No `db.query(...)` in
route functions for non-trivial queries.

SQLAlchemy 2.0 style: `select(User).where(...)`, `await db.scalar(...)`,
`await db.scalars(...).all()`. No legacy `db.query(User).filter_by(...)`.

Migrations via Alembic. No `Base.metadata.create_all()` in production
code paths — only in test fixtures.

Before — module-level session, inline query, no transaction:

db = SessionLocal()

@app.post("/users")
def create(data: dict):
    u = User(email=data["email"])
    db.add(u)
    db.commit()  # commits on a shared session across requests — race city
    return u

After — request-scoped async session, repository, explicit transaction:

async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

@router.post("/users", response_model=UserRead, status_code=201)
async def create_user(
    body: UserCreate,
    db: AsyncSession = Depends(get_db),
) -> UserRead:
    async with db.begin():
        user = await users_repo.create(db, body)
    return UserRead.model_validate(user)

# users_repo.py
async def create(db: AsyncSession, body: UserCreate) -> User:
    user = User(email=body.email, password_hash=hash_pw(body.password), name=body.name)
    db.add(user)
    await db.flush()
    return user

One session per request. Transaction committed or rolled back atomically.

---

Rule 4: Dependencies for Cross-Cutting Concerns — Auth, Rate Limits, Context

Cursor routinely puts authentication, authorization, rate limiting, and tenant-context resolution inside handlers — if request.headers.get("authorization") != f"Bearer {token}": raise .... FastAPI's dependency system is built for exactly this: compose dependencies, declare them on routers and endpoints, test them in isolation. The result is handlers that read like business logic, not middleware.

The rule:

Every cross-cutting concern is a dependency:
  - `get_current_user` (reads JWT, returns User or raises 401)
  - `require_scope("users:write")` (checks claim, raises 403)
  - `rate_limit("user_create", per_minute=10)` (slowapi / fastapi-limiter)
  - `get_tenant` (resolves tenant from host/header)
  - `get_request_id` (reads X-Request-Id or generates one)

Handlers declare them explicitly:
  async def create_user(
      body: UserCreate,
      user: User = Depends(get_current_user),
      _: None = Depends(require_scope("users:write")),
      db: AsyncSession = Depends(get_db),
  ) -> UserRead: ...

Auth schemes use FastAPI's built-ins (OAuth2PasswordBearer,
HTTPBearer, APIKeyHeader). Never parse `Authorization` manually.

Router-level dependencies for things that apply everywhere:
  admin_router = APIRouter(
      prefix="/admin",
      dependencies=[Depends(get_current_user), Depends(require_admin)],
  )

Dependencies returning data (current_user, tenant) are cached per-request
by default — FastAPI handles this. Don't cache manually.

Never put logic in middleware that needs per-route configuration —
middleware is for truly global concerns (CORS, GZip, request logging).
Per-route lives in dependencies.

Before — auth parsed inline, no scope check, handler mixes concerns:

@app.post("/users")
def create(req: Request, data: UserCreate, db: Session = Depends(get_db)):
    auth = req.headers.get("authorization") or ""
    token = auth.removeprefix("Bearer ").strip()
    if not token:
        raise HTTPException(401)
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    if "admin" not in payload.get("scopes", []):
        raise HTTPException(403)
    # ... create user ...

After — composed dependencies, declarative:

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

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: AsyncSession = Depends(get_db),
) -> User:
    try:
        payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
    except JWTError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")
    user = await users_repo.get(db, UUID(payload["sub"]))
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found")
    return user

def require_scope(scope: str) -> Callable[..., None]:
    async def checker(token: Annotated[str, Depends(oauth2_scheme)]) -> None:
        payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
        if scope not in payload.get("scopes", []):
            raise HTTPException(status.HTTP_403_FORBIDDEN, f"missing scope {scope}")
    return checker

@router.post("/users", response_model=UserRead, status_code=201)
async def create_user(
    body: UserCreate,
    user: Annotated[User, Depends(get_current_user)],
    _: Annotated[None, Depends(require_scope("users:write"))],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> UserRead:
    ...

Testing auth is one-liners: override get_current_user in a test fixture.

---

Rule 5: Structured Error Handling — Typed Exceptions, Problem-Details Responses

Cursor's default error path is try: ... except Exception as e: return {"error": str(e)}, 500. That leaks internals, has no machine-readable error code, and swallows the stack trace so nobody can debug. FastAPI's exception handlers plus RFC-7807 (Problem Details) give you consistent, typed error responses the frontend can switch on.

The rule:

Domain errors are typed subclasses of `AppError` (or `HTTPException` for
wire-level). Never raise bare `Exception` or return `{"error": "..."}`.

class AppError(Exception):
    status_code: int
    code: str  # machine-readable: "user_not_found", "invalid_otp"
    message: str

Common subclasses: NotFoundError, ConflictError, ValidationError,
AuthenticationError, AuthorizationError, RateLimitError.

A global exception handler translates them to Problem Details JSON:
  {
    "type": "https://example.com/errors/user-not-found",
    "title": "User not found",
    "status": 404,
    "code": "user_not_found",
    "request_id": "...",
    "detail": "User with id 42 does not exist"
  }

Unhandled `Exception` goes through a catch-all handler that logs with
stack trace and returns a generic 500 — never leaks `str(e)`.

ValidationError from Pydantic stays as 422 with the default FastAPI
shape; don't rewrite it.

Logging uses structlog or stdlib logging with contextvars for
request_id / user_id. Never `print`. Every exception handler logs
the full exception with `exc_info=True`.

Before — bare Exception, leaks error string, 200 with error body:

@app.get("/users/{id}")
def get_user(id: int):
    try:
        return db.query(User).get(id)
    except Exception as e:
        return {"error": str(e)}

After — typed errors, structured response, logged:

class AppError(Exception):
    status_code = 500
    code = "internal_error"
    def __init__(self, message: str, *, detail: str | None = None):
        super().__init__(message); self.message = message; self.detail = detail

class NotFoundError(AppError):
    status_code = 404
    code = "not_found"

@app.exception_handler(AppError)
async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
    log.warning("app_error", code=exc.code, detail=exc.detail, request_id=get_request_id())
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "type": f"https://errors.example.com/{exc.code}",
            "title": exc.message,
            "status": exc.status_code,
            "code": exc.code,
            "request_id": get_request_id(),
            "detail": exc.detail,
        },
    )

@app.exception_handler(Exception)
async def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
    log.error("unhandled_exception", exc_info=True, request_id=get_request_id())
    return JSONResponse(500, content={"type": "about:blank", "title": "Internal Server Error", "status": 500, "code": "internal_error", "request_id": get_request_id()})

@router.get("/users/{user_id}", response_model=UserRead)
async def get_user(user_id: UUID, db: AsyncSession = Depends(get_db)) -> UserRead:
    user = await users_repo.get(db, user_id)
    if user is None:
        raise NotFoundError("User not found", detail=f"User {user_id} does not exist")
    return UserRead.model_validate(user)

Consistent wire shape. Secrets never leak. Every error logged with request_id.

---

Rule 6: BackgroundTasks vs Workers — Know When Not to Use BackgroundTasks

FastAPI's BackgroundTasks is perfect for "fire-and-forget after the response is sent" when the task is small, in-process, and can't afford a dedicated worker. Cursor reaches for it when what you actually need is a durable background job — email send with retries, invoice PDF generation, webhook delivery. Those require a real worker (Celery, Arq, Dramatiq, RQ) or cloud-managed queues (SQS + Lambda). Put them in BackgroundTasks and they die with the process.

The rule:

Use `BackgroundTasks` ONLY when all of these are true:
  - The work is seconds, not minutes
  - Losing the task on process restart is acceptable
  - The task has no retry semantics
  - The task doesn't hold resources (DB connections, external client handles)

Examples: audit log write, cache invalidation, nonessential metrics ping.

Use a real worker (Arq for async, Celery for sync, Dramatiq for either)
when any of:
  - The task can fail and must retry
  - It's durable (must survive deploy)
  - It's slow (>5s)
  - It fans out to external APIs with their own rate limits
  - It needs scheduling / cron

Examples: email send, webhook delivery, PDF/image generation,
import/export, LLM calls.

Never share a `BackgroundTasks` instance across requests. Declare it
as a dependency in the handler signature:
  background_tasks: BackgroundTasks

Background-task functions must be testable: accept all dependencies
as arguments, never read from `app.state` inside.

Workers consume the same Pydantic schemas (`UserRead`, `OrderCreate`)
— never re-serialize with ad-hoc JSON shapes.

Before — BackgroundTasks for an email with retries, no observability:

@app.post("/signup")
async def signup(body: SignupBody, bg: BackgroundTasks):
    user = await create(body)
    bg.add_task(send_welcome_email, user.email)
    return {"ok": True}

def send_welcome_email(email: str):
    smtp.send(email, "Welcome!", "...")  # fails silently, never retries

After — Arq worker for durable send, BackgroundTasks only for the audit log:

@router.post("/signup", response_model=SignupResponse, status_code=201)
async def signup(
    body: SignupBody,
    bg: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    arq: ArqRedis = Depends(get_arq),
) -> SignupResponse:
    user = await users_repo.create(db, body)

    await arq.enqueue_job("send_welcome_email", user.id, _job_id=f"welcome:{user.id}")
    bg.add_task(write_audit, "user.signup", user.id)

    return SignupResponse(id=user.id)

# worker/tasks.py
async def send_welcome_email(ctx: Context, user_id: UUID) -> None:
    async with ctx["db"]() as db:
        user = await users_repo.get(db, user_id)
    await ctx["mailer"].send(user.email, "welcome", {"name": user.name})

The email retries on failure with exponential backoff, survives deploy, is idempotent via _job_id. The audit log uses BackgroundTasks because losing it on restart is acceptable.

---

Rule 7: Router Layout, Versioning, and OpenAPI Discipline

The default FastAPI project Cursor scaffolds has one main.py with forty endpoints. That's fine for a tutorial. It's not shippable. Routers per resource, versioned prefixes, and hand-curated OpenAPI metadata make the API legible to SDK generators, frontend engineers, and the humans debugging it.

The rule:

Module layout:
  app/
    main.py              # FastAPI() + lifespan + middleware wiring
    api/
      __init__.py
      deps.py            # get_db, get_current_user, ...
      v1/
        __init__.py      # v1_router = APIRouter(prefix="/v1"); include sub-routers
        users.py         # router = APIRouter(prefix="/users", tags=["users"])
        orders.py
      v2/
    core/
      config.py          # pydantic-settings BaseSettings
      security.py        # JWT, hashing
    db/
      session.py
      base.py
    models/              # SQLAlchemy ORM
    schemas/             # Pydantic
    repositories/
    services/

Versioning in the URL prefix. Internal-only changes bump the OpenAPI
version, not the URL. Breaking changes go to v2.

Every endpoint has:
  - `summary` (short, used as operationId)
  - `description` (one paragraph, rendered in Swagger)
  - `response_model`
  - `responses={404: {"model": ProblemDetails}, 409: ...}` for non-default codes
  - `status_code` explicit on non-200 success

`tags=["..."]` on every router for grouped docs.

Generate SDKs from the OpenAPI spec (`openapi-typescript-codegen`,
`openapi-generator`), not hand-written TypeScript. If generation
fails, the OpenAPI description is wrong — fix it.

Never ship routes tagged "test" or "debug" to production. Guard them
with an env check or a dependency that 404s unless `settings.env == "local"`.

Before — one main.py, no versioning, no response_model:

app = FastAPI()

@app.get("/users/{id}")
def get_user(id: int): ...

@app.post("/users")
def create_user(data: dict): ...

@app.get("/orders/{id}")
def get_order(id: int): ...

After — versioned router tree with documented responses:

# app/api/v1/users.py
router = APIRouter(prefix="/users", tags=["users"])

@router.get(
    "/{user_id}",
    response_model=UserRead,
    summary="Get a user by id",
    description="Returns a single user. 404 if the user does not exist.",
    responses={404: {"model": ProblemDetails, "description": "User not found"}},
)
async def get_user(user_id: UUID, db: AsyncSession = Depends(get_db)) -> UserRead:
    ...

# app/api/v1/__init__.py
v1_router = APIRouter(prefix="/v1")
v1_router.include_router(users.router)
v1_router.include_router(orders.router)

# app/main.py
app = FastAPI(
    title="Widget API",
    version="1.4.0",
    lifespan=lifespan,
)
app.include_router(v1_router)

Swagger is navigable. SDK generation works on the first try.

---

Rule 8: Testing With pytest-asyncio and httpx.AsyncClient — Real HTTP Against a Real ASGI

Cursor's default FastAPI test is client = TestClient(app); client.get("/users/1") — which works, but hides async issues because TestClient under the hood pumps a sync wrapper around the ASGI app. For anything that exercises the real async path (streaming, websockets, true concurrency, middleware interactions), use httpx.AsyncClient with the ASGITransport. For DB, use an ephemeral database fixture, not mocks.

The rule:

Test runner: pytest + pytest-asyncio with `asyncio_mode = "auto"`.

HTTP tests use httpx.AsyncClient with ASGITransport:
  async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
      resp = await client.get("/v1/users/42")

Database: either
  - pytest fixture that creates a fresh schema per test (fastest on SQLite :memory:)
  - pytest-docker + Postgres for parity with production (required if you
    use pgvector, JSONB operators, CTEs, etc.)
Apply migrations in the fixture via Alembic upgrade head — don't
Base.metadata.create_all() and drift from production DDL.

Override dependencies in tests via app.dependency_overrides:
  app.dependency_overrides[get_current_user] = lambda: test_user
Always clean up in the teardown: app.dependency_overrides.clear().

Factories (factory_boy / polyfactory) for schema test data — never
hand-type 50-line dicts.

Test the wire, not the handler. Assert on the JSON response shape
and status. Never call the handler function directly — you bypass
dependencies, validation, serialization.

External services (Stripe, S3, OpenAI): respx for httpx (or aioresponses)
to stub at the HTTP boundary. Never replace the client with a MagicMock
that doesn't match the real client's interface.

Coverage gates: >85% on services/repositories, >70% on routers,
mutation testing (mutmut, cosmic-ray) for critical billing / auth paths.

Before — TestClient, handler called directly, mocked at repo layer:

def test_create_user():
    result = create_user(UserCreate(email="a@b.c", name="A", password="x" * 12), db=MagicMock())
    assert result.email == "a@b.c"

Bypasses every dependency, every validation, every middleware.

After — real ASGI round-trip, fresh DB, respx for outbound:

@pytest_asyncio.fixture
async def client(app_with_db) -> AsyncIterator[AsyncClient]:
    transport = ASGITransport(app=app_with_db)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

@pytest.mark.asyncio
async def test_create_user_returns_201_and_strips_password(client: AsyncClient, auth_headers):
    resp = await client.post(
        "/v1/users",
        json={"email": "[e-mail]", "name": "Ada", "password": "correct horse battery staple"},
        headers=auth_headers,
    )
    assert resp.status_code == 201
    body = resp.json()
    assert body["email"] == "[e-mail]"
    assert "password" not in body
    assert "password_hash" not in body

@pytest.mark.asyncio
async def test_create_user_rejects_extra_field(client: AsyncClient, auth_headers):
    resp = await client.post(
        "/v1/users",
        json={"email": "a@b.c", "name": "A", "password": "x" * 12, "role": "admin"},
        headers=auth_headers,
    )
    assert resp.status_code == 422

@pytest.mark.asyncio
async def test_signup_enqueues_welcome_email(client: AsyncClient, arq_inspector):
    resp = await client.post("/v1/signup", json={"email": "a@b.c", "name": "A", "password": "x" * 12})
    assert resp.status_code == 201
    jobs = await arq_inspector.jobs("send_welcome_email")
    assert len(jobs) == 1
    assert jobs[0].args[0] == UUID(resp.json()["id"])

Exercises the real stack. Catches the password_hash leak. Verifies the side effect (job enqueued) via the queue's own inspector.

---

The Complete .cursorrules File

Drop this in the repo root. Cursor and Claude Code both pick it up.

# FastAPI — Production Patterns

## Pydantic v2 Schemas
- Three schemas per resource: {X}Base / {X}Create / {X}Update / {X}Read.
  Never return an ORM model directly.
- ConfigDict(extra="forbid", strict=True, str_strip_whitespace=True);
  from_attributes=True only when reading from ORM.
- Every endpoint declares `response_model=...` and a typed return
  annotation. No `dict` / `Any`.
- Pagination is a typed `Page[T]` with items/total/page/size — never
  list+count pair.
- Field validation via Annotated[..., Field(...)] and @field_validator;
  no ad-hoc `if not x: raise` in handlers.

## Async Discipline
- New routers are `async def`; legacy sync routers explicitly `def`.
  Never mixed.
- Blocking libraries (requests, psycopg2, sync redis) banned from async
  handlers. Use httpx.AsyncClient, asyncpg/SQLAlchemy AsyncSession,
  redis.asyncio.
- Unavoidable blocking: asyncio.to_thread.
- One httpx.AsyncClient per app, managed by `lifespan`.
- `time.sleep` forbidden in request path.
- CPU-bound work in ProcessPoolExecutor or a background worker.

## Database
- `AsyncSession` via `get_db` dependency that yields and closes.
- engine with pool_pre_ping=True, pool_size, async_sessionmaker with
  expire_on_commit=False.
- Explicit transactions via `async with db.begin():`.
- Queries live in repositories; handlers orchestrate, they don't build SQL.
- SQLAlchemy 2.0 style (select(), db.scalars) only.
- Alembic for migrations; no create_all() in production paths.

## Dependencies
- Every cross-cutting concern is a dependency (auth, scopes, rate limit,
  tenant, request_id).
- Auth via OAuth2PasswordBearer / HTTPBearer / APIKeyHeader — never
  manual Authorization parsing.
- Router-level dependencies for shared concerns (admin router).
- Middleware only for truly global (CORS, GZip, request logging);
  per-route lives in deps.

## Errors
- Typed AppError subclasses (NotFoundError, ConflictError, ...). No bare
  Exception. No {"error": str(e)}.
- Global exception handler produces Problem Details JSON with request_id.
- Unhandled Exception logged with exc_info=True; response is generic 500.
- Pydantic ValidationError stays 422 with default shape.
- Logging via structlog or stdlib with contextvars; no print.

## Background Work
- BackgroundTasks only for in-process, fail-ok, sub-second work.
- Real worker (Arq/Celery/Dramatiq) for retries, durability, slow,
  external-fanout, scheduled.
- Background-task functions take all deps as args; never read app.state.
- Workers use the same Pydantic schemas as the API.

## Layout & OpenAPI
- app/api/v{N}/{resource}.py routers with prefix+tags; main.py wires lifespan.
- Versioning in URL prefix; breaking changes bump major.
- Every endpoint: summary, description, response_model, `responses`
  for non-default codes, explicit status_code.
- Generate SDKs from OpenAPI; hand-written clients banned.
- Debug/test endpoints guarded by `settings.env == "local"`.

## Testing
- pytest + pytest-asyncio (asyncio_mode=auto).
- httpx.AsyncClient with ASGITransport; no TestClient for new tests.
- Fresh DB per test (SQLite memory or dockerized Postgres); apply
  migrations via Alembic, not create_all.
- `app.dependency_overrides` for auth/db swaps, cleared in teardown.
- Factories (factory_boy / polyfactory) for test data.
- respx / aioresponses at the HTTP boundary; no MagicMock on clients.
- Coverage >85% services, >70% routers; mutation testing on critical paths.

---

End-to-End Example: A Paginated List Endpoint With Filtering

Without rules: dict return, inline session, manual auth, offset-based pagination, no response_model.

@app.get("/users")
def list_users(q: str = None, page: int = 1, request: Request = None):
    if "Bearer " not in (request.headers.get("authorization") or ""):
        return {"error": "auth"}, 401
    db = SessionLocal()
    query = db.query(User)
    if q:
        query = query.filter(User.name.ilike(f"%{q}%"))
    total = query.count()
    users = query.offset((page - 1) * 20).limit(20).all()
    return {"users": [u.__dict__ for u in users], "total": total}

With rules: typed Page, repository, dependency-injected auth and DB, keyset pagination.

class UserFilter(BaseModel):
    model_config = ConfigDict(extra="forbid")
    q: str | None = Field(None, min_length=1, max_length=100)
    cursor: UUID | None = None
    limit: Annotated[int, Field(ge=1, le=100)] = 20

@router.get("/users", response_model=Page[UserRead], summary="List users")
async def list_users(
    filters: Annotated[UserFilter, Query()],
    user: Annotated[User, Depends(get_current_user)],
    _: Annotated[None, Depends(require_scope("users:read"))],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> Page[UserRead]:
    page = await users_repo.list_keyset(db, filters)
    return Page[UserRead](
        items=[UserRead.model_validate(u) for u in page.items],
        next_cursor=page.next_cursor,
        size=filters.limit,
    )

Typed input, typed output, keyset pagination that scales past a million rows, auth/scope/DB all injected.

---

Get the Full Pack

These eight rules cover the FastAPI patterns where AI assistants consistently reach for the wrong idiom. Drop them into .cursorrules and the next prompt you write will look different — strict-schema, async-disciplined, session-scoped, dependency-injected, typed-error, worker-routed, versioned, httpx-tested FastAPI, without having to re-prompt.

If you want the expanded pack — these eight plus rules for OpenAPI contract testing, uvicorn/gunicorn tuning, OpenTelemetry instrumentation, Alembic migration patterns, streaming responses and server-sent events, websocket discipline, and the testing conventions I use on production FastAPI services — it is bundled in Cursor Rules Pack v2 ($27, one payment, lifetime updates). Drop it in your repo, stop fighting your AI, ship FastAPI you would actually merge.