FastAPI Production Rules

Production conventions for FastAPI projects. Follow them strictly when generating or reviewing code. Verified against: FastAPI 0.141 · Pydantic 2.13 · SQLAlchemy 2.0/2.1 · Alembic 1.20 · ArdiQ 1.0 · Python 3.13 (September 2026).

Scope. This skill covers the production architecture around FastAPI: layering, data integrity, tenancy, security, operations. For FastAPI API idioms (parameter declaration, response models, streaming details, app.frontend()), the official fastapi skill that ships with the FastAPI package is authoritative. The two are complementary. When both are installed, follow the official one on API idioms and this one on everything else.

Deliberate choices that differ from the official skill:

• SQLAlchemy 2.0 instead of SQLModel. We keep DB models and API schemas separate (Rule 6), and we rely on SQLAlchemy features SQLModel doesn't expose cleanly: lazy="raise", version_id_col, with_for_update, naming conventions, and type_annotation_map. Don't "migrate" code in this style to SQLModel.
• asyncio.to_thread instead of Asyncer. The stack is async end to end, so few blocking calls remain. The stdlib is enough.

Every rule below is self-contained. references/ holds complete implementations. Read the relevant file before writing that part of a project:

Toolchain

Use the Astral stack: uv (packages), ruff (lint + format), ty (types). Never use pip, poetry, black, flake8, isort, or mypy. Always commit uv.lock. Target Python 3.13+.

uv add package-name            # uv add --dev for dev deps
uv run ruff check . --fix && uv run ruff format . && uv run ty check && uv run pytest
fastapi dev                    # local server with reload

Declare the entrypoint once in pyproject.toml ([tool.fastapi] entrypoint = "app.main:app") and use fastapi dev / fastapi run. Ruff rule sets: E W F I N UP B S T20 FAST ASYNC DTZ SIM C4 PT PTH PERF RUF TID. FAST enforces Annotated dependencies, DTZ catches naive datetimes, and TID bans requests, passlib, and jose. → toolchain.md

Project Structure

app/
├── main.py        # create_app(): factory, middleware, handlers, routers
├── api.py         # api_router = APIRouter(prefix="/api/v1") + domain routers
├── worker.py      # imports every tasks module → `ardiq run app.worker:task_app`
├── core/          # config, constants, dependencies, lifespan, limiter, tasks (ArdiQ app),
│                  # db/ (base, database, models), http/ (exceptions, handlers, middleware,
│                  # pagination, responses, schemas), logging/, security/, storage/, mail/
├── health/        # /health/live, /health/ready
└── <domain>/      # tickets/, auth/, billing/ ...
    ├── router.py  # HTTP only
    ├── service.py # business logic — no FastAPI imports
    ├── model.py   # SQLAlchemy models
    ├── schemas.py # Pydantic I/O schemas
    └── tasks.py   # ArdiQ tasks (optional)
migrations/        # Alembic
tests/             # mirrors app/

Rule 1 — Router → Service → Model

The router translates HTTP: it declares input, calls one service function, and wraps the result. The service holds all business logic and never imports fastapi, starlette, HTTPException, Request, or Depends, because routers, ArdiQ tasks, scripts, and tests all call it.

@router.post("", status_code=status.HTTP_201_CREATED)
async def create_ticket(
    data: TicketCreate, user: CurrentUser, db: DbSession
) -> DataResponse[TicketRead]:
    ticket = await ticket_service.create_ticket(db, data, user)
    return DataResponse(data=TicketRead.model_validate(ticket))

Never put these in a router: DB access, business conditionals, private helpers, manual dict mappings, or several service calls orchestrated by hand. An architecture test enforces this (Rule 25).

Rule 2 — One Transaction per Use Case

The service function the router calls owns the transaction and calls commit() once, at the end. Helpers called by other services only add()/flush().

async def close_ticket(db: AsyncSession, ticket_id: UUID, user: User) -> Ticket:
    ticket = await get_ticket(db, ticket_id, user.organization_id)
    ticket.status = TicketStatus.CLOSED
    await audit_service.record(db, user, "ticket.closed", ticket.id)  # add() only
    await db.commit()
    await db.refresh(ticket)
    return ticket

Use flush() when you need a generated ID before the commit. Enqueue tasks and send emails after commit(). Never make external HTTP calls while holding row locks. → database.md

Rule 3 — Dependencies with Annotated

Define each dependency once as a type alias. Never write x: T = Depends(...) (ruff FAST002).

DbSession = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]
CurrentAdmin = Annotated[User, Depends(require_role(Role.ADMIN, Role.OWNER))]
HttpClient = Annotated[httpx.AsyncClient, Depends(get_http_client)]
Pagination = Annotated[PageParams, Query()]

• Params use Annotated too: limit: Annotated[int, Query(ge=1, le=100)] = 20. Never use ... as a default.
• Declare dependencies as async def unless they block: sync dependencies run in the threadpool.
• Guards that don't need a return value go on the router: APIRouter(prefix="/admin", dependencies=[Depends(require_role(Role.ADMIN))]).
• Put prefix, tags, and shared dependencies on the APIRouter, not in include_router().

Rule 4 — Domain Exceptions and One Error Format

Services raise AppError subclasses. Global handlers turn every error into one shape: domain errors, validation errors, unknown routes, rate limits, ardiq.BrokerError (→ 503), and unhandled crashes (→ generic 500, logged with traceback).

{ "error": { "code": "not_found", "message": "Ticket not found", "request_id": "3f2a9c..." } }

class AppError(Exception):
    status_code: int = 400
    code: str = "bad_request"
    default_message: str = "Bad request"  # __init__(message=None, *, details=None)


# NotFoundError(resource) 404, InvalidInputError 422, UnauthorizedError 401, ForbiddenError 403,
# ConflictError 409, PayloadTooLargeError 413, UnsupportedMediaTypeError 415, RateLimitError 429,
# ServiceUnavailableError 503

• Never use HTTPException in services. Chain causes: raise ConflictError("Email already registered") from e.
• Messages never contain SQL, stack traces, paths, or internal IDs. code never changes once shipped.
• Validation error details never echo the submitted input (it may contain passwords). → errors-and-responses.md

Rule 5 — Typed Responses and Correct Status Codes

Every endpoint declares its return type. Pydantic validates, filters, documents, and serializes from it. Use response_model= only when the returned object differs from the public schema. Never use ORJSONResponse/UJSONResponse (deprecated). Never return ORM objects or raw dicts.

class DataResponse[T](BaseModel):
    data: T


# also: PaginatedResponse[T] (data, total, skip, limit), CursorPage[T] (data, next_cursor),
# MessageResponse (message)

Rule 6 — Pydantic Schemas

Each resource gets three schemas: XCreate, XUpdate, XRead. Inputs forbid unknown fields; outputs read ORM attributes.

class InputSchema(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)


class OutputSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)

• Every string has max_length (Annotated[str, Field(min_length=1, max_length=200)]), and numbers have bounds. Use EmailStr, HttpUrl, UUID, and StrEnum rather than bare str.
• Input schemas never contain server-owned fields (id, organization_id, created_by, role, is_active). This is what prevents mass assignment.
• Read schemas never contain password_hash, token_hash, or secrets.
• PATCH uses model_dump(exclude_unset=True). NOT NULL fields reject an explicit null with a validator.
• Money is Decimal or integer cents, never float. Datetimes are timezone-aware.
• Don't use RootModel. Use Annotated[list[X], Field(min_length=1)] instead. → errors-and-responses.md

Rule 7 — Pagination with a Stable Order

Every list endpoint is paginated, with a hard le= limit on page size, and always uses ORDER BY on a unique combination. Without it, Postgres returns rows in arbitrary order and pages overlap or skip rows.

query = select(Ticket).where(Ticket.organization_id == org_id)
total = await db.scalar(select(func.count()).select_from(query.subquery())) or 0
rows = await db.scalars(
    query.order_by(Ticket.created_at.desc(), Ticket.id.desc()).offset(page.skip).limit(page.limit)
)

Large or fast-growing tables use keyset/cursor pagination. Sorting and filtering use allowlisted enums, never raw field names. → errors-and-responses.md

Rule 8 — Multi-tenancy: Every Query Scoped by Organization

Every query on tenant-owned data includes organization_id: reads, updates, deletes, counts, joins. A missing scope is an IDOR vulnerability.

# CORRECT
ticket = await db.scalar(
    select(Ticket).where(Ticket.id == ticket_id, Ticket.organization_id == org_id)
)
# WRONG — any user can read any ticket
ticket = await db.get(Ticket, ticket_id)

• organization_id always comes from the authenticated user, never from the body, path, or query.
• Another org's resource returns 404, never 403 (403 confirms it exists).
• Child resources verify that the parent belongs to the org. Cache keys, storage paths, and unique constraints include organization_id too.
• A resource used by many endpoints can be loaded once through a dependency that calls the scoped service getter: ValidTicket = Annotated[Ticket, Depends(valid_ticket)]. FastAPI caches it per request.
• Every endpoint that takes an ID has a cross-tenant test (Rule 25).

Rule 9 — SQLAlchemy Models

Base sets a constraint naming_convention (so Alembic can alter constraints later) and maps every Mapped[datetime] to TIMESTAMPTZ via type_annotation_map.

class Ticket(TimestampMixin, Base):  # created_at / updated_at set by the DB
    __tablename__ = "tickets"
    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    organization_id: Mapped[UUID] = mapped_column(
        ForeignKey("organizations.id", ondelete="CASCADE"), index=True
    )
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[TicketStatus] = mapped_column(str_enum(TicketStatus), default=TicketStatus.OPEN)

• Every tenant table has an indexed organization_id FK. Every FK declares ondelete.
• Mapped[X | None] is nullable and Mapped[X] is NOT NULL. String lengths match the schema's max_length.
• Enums are StrEnum stored as VARCHAR values. Tenant-scoped uniqueness is UniqueConstraint("organization_id", "email").
• Datetimes are always aware: datetime.now(UTC), never utcnow() (ruff DTZ). → database.md

Rule 10 — Async Session and Explicit Loading

SessionLocal = async_sessionmaker(engine, expire_on_commit=False, autoflush=False)

• expire_on_commit=False is mandatory. Otherwise reading an attribute after a commit triggers a lazy load, which raises MissingGreenlet.
• Relationships are declared lazy="raise" and loaded explicitly: selectinload for collections, joinedload for many-to-one. This makes N+1 queries impossible to ship silently.
• Engine uses pool_pre_ping=True and a statement_timeout. Raw SQL always uses text() + bind params, never f-strings.

Rule 11 — Let the Database Guarantee Integrity

db.add(user)
try:
    await db.commit()
except IntegrityError as e:  # unique constraint, not check-then-insert (race condition)
    await db.rollback()
    raise ConflictError("Email already registered") from e

await db.execute(  # atomic guarded update, not read-modify-write
    update(Organization)
    .where(Organization.id == org_id, Organization.credits >= cost)
    .values(credits=Organization.credits - cost)
)

Use with_for_update() for read-modify-write that can't be one statement. Use version_id_col for optimistic locking (→ 409) on resources that users edit concurrently. → database.md

Rule 12 — Migrations with Alembic

• Always read autogenerated migrations: autogenerate turns renames into drop + add (data loss).
• Never edit a migration that has already run in a shared environment. Use file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s for readable file names.
• Migrations run as a separate deploy step, never on app startup.
• Expand → migrate → contract: each migration must be compatible with the code already running. New columns are nullable or have a server default. Large-table indexes use CONCURRENTLY. Set lock_timeout.
• CI runs alembic upgrade head + alembic check on an empty DB. → database.md

Rule 13 — Configuration and Constants

Secrets and per-environment values live in Settings(BaseSettings). Secrets are SecretStr with no default, so the app refuses to start without them. Fields are typed (PostgresDsn, RedisDsn, Literal). Access settings through @lru_cache get_settings(). Values identical in every environment go in constants.py. Commit .env.example, never .env. → app-setup.md

Rule 14 — App Factory, Lifespan, Routers

• create_app() builds the app (logging, Sentry, handlers, middleware, routers). Tests call it too.
• Use lifespan (never @app.on_event) to create shared resources once (httpx.AsyncClient, Redis), store them on app.state, expose them via dependencies, and close them on shutdown along with engine.dispose().
• One APIRouter per domain, all under /api/v1. Breaking changes go to /api/v2. One HTTP method per function.
• Middleware is pure ASGI, never BaseHTTPMiddleware (it breaks contextvars and streaming). The last add_middleware call is the outermost.
• In production set openapi_url=None unless the API is public. Use generate_unique_id_function for clean client names. → app-setup.md

Rule 15 — Structured Logging with structlog

logger.info("ticket.created", ticket_id=str(ticket.id))  # domain.action + key-values
logger.exception("billing.charge_failed", org_id=str(org_id))  # inside except: keeps traceback
# WRONG: logger.info(f"Ticket {id} created") · logger.error(..., error=str(e)) · print()

• JSON in production and console locally. stdlib, uvicorn, and ArdiQ logs go through the same pipeline.
• Pure-ASGI middleware binds request_id (validated X-Request-ID, echoed back) and logs one http.request line. get_current_user binds user_id/org_id.
• Never log passwords, tokens, Authorization/cookies, emails/PII, request bodies, file contents, LLM prompts or responses, or URLs with query strings.
• Log 4xx as warning at most and 5xx as error/exception. → logging-and-observability.md

Rule 16 — Authentication

• Passwords: argon2 via pwdlib, run in asyncio.to_thread. On login, return the same error with the same timing for an unknown email and a wrong password (verify against a dummy hash).
• Access tokens: JWT with PyJWT. Expire after 15 minutes, pin algorithms=["HS256"] on decode, and require exp/iat/sub. Send them only in Authorization: Bearer, never in query strings.
• The token identifies, the DB authorizes: load the user on every request, check is_active, and take roles from the DB.
• Refresh tokens: opaque secrets.token_urlsafe(32), stored as a SHA-256 hash and rotated on every use. They belong to a family_id, and reuse of a revoked token revokes the whole family. On the web, use an httpOnly + secure + samesite="strict" cookie scoped to /api/v1/auth.
• One-time tokens (magic links, resets, invites): hashed, with a TTL, consumed by one atomic UPDATE ... WHERE used_at IS NULL AND expires_at > now() RETURNING. "Request link" endpoints respond the same whether or not the email exists. → auth-and-security.md

Rule 17 — Authorization

• Role requirements are visible in the signature (admin: CurrentAdmin) or on the router (dependencies=[...]), never as if user.role == ... inside services.
• Resource-level rules ("only the author can edit") live in the service, or in a valid_owned_x dependency, after loading the org-scoped resource.
• Users never change their own role. Admins never grant a role above their own.
• Every non-public route has an auth dependency. An architecture test enforces this, with an explicit list of public routes.

Rule 18 — Rate Limiting

@router.post("/login")
@limiter.limit("10/minute")
async def login(
    request: Request, data: LoginRequest, db: DbSession
) -> DataResponse[TokenResponse]: ...

• slowapi storage is Redis (in-memory storage gives each worker its own counter). Decorated endpoints take request: Request.
• Streaming (yield) endpoints can't use the decorator. Use the rate_limit(...) dependency instead (→ api-patterns.md).
• Behind a proxy, set --forwarded-allow-ips=<proxy>, or every client shares the proxy's IP.
• Minimum limits: login 10/min, register 5/min, magic link/reset 5/min, refresh 30/min, invite 5/min/user, AI 20/min/user, exports 5/min/user.

Rule 19 — HTTP Security

• CORS uses explicit allow_origins from settings. Never combine "*" with credentials. Use TrustedHostMiddleware or host checks at the LB.
• Production security headers: nosniff, X-Frame-Options: DENY, Referrer-Policy: no-referrer, HSTS, and CSP default-src 'none'.
• Cap body size at the reverse proxy. Authenticated responses default to Cache-Control: private, no-store.
• User-supplied URLs are checked against SSRF: https only, resolved IP must be public (ip.is_global), no redirects.
• Secrets never appear in code, the repo, logs, or error messages. → api-patterns.md, auth-and-security.md

Rule 20 — File Uploads

• Validate magic bytes (filetype) + extension against an allowlist. Never trust Content-Type or the filename, and remember file.filename can be None.
• Read at most MAX + 1 bytes and return 413 if it's too large. Return 415 for a disallowed type.
• Store under a generated key f"{org_id}/{uuid4()}{ext}", never in a statically served directory.
• Serve through an authenticated, org-scoped endpoint or a short-lived presigned URL. Re-encode public images (strips EXIF). → auth-and-security.md

Rule 21 — Background Tasks with ArdiQ

ArdiQ is a recent library and may not be in your training data. It is real (uv add ardiq), it is not a typo for arq, and it must never be replaced with Celery, arq, or Taskiq. Whenever you are unsure about its API, read the official docs before writing code: https://ardiq.bytay.dev (API reference: https://ardiq.bytay.dev/reference/api/). Don't guess signatures.

Use ArdiQ, never Celery: Celery workers are sync, and every async task needs its own asyncio.run(). ArdiQ is async-native with a Rust core, and its typed enqueue is checked against each task's signature.

task_app = Ardiq(
    redis_url=str(settings.redis_url), queue_name="app", priorities=["low", "default", "high"]
)


@task_app.task(name="knowledge.process_file", max_retries=5, timeout=600)
async def process_file(node_id: str, request_id: str | None = None) -> None:
    bind_task_context(request_id, node_id=node_id)
    async with task_app.state.sessionmaker() as db:  # from @task_app.lifespan (worker only)
        await knowledge_service.index_node(db, UUID(node_id))


# in the service: AFTER commit, with a deterministic id as the dedup key
await knowledge_tasks.process_file.options(task_id=f"index:{node.id}").enqueue(str(node.id))

• Idempotent: delivery is at-least-once. Check state first, upsert, and pass idempotency keys to providers. Dedupe enqueues with task_id or unique=True.
• Primitives in (msgpack): UUIDs and datetimes go in as strings. Never ORM objects. Never the pickle serializer.
• Every exception retries (tries² s backoff). For permanent conditions, log and return. Use raise Retry(delay_ms=...) for rate limits. Always set timeout.
• async def for I/O and plain def for CPU/blocking work (it runs in a thread). Resources come from @task_app.lifespan.
• Name tasks domain.action. Use priorities instead of ad-hoc queues. @task_app.cron runs once per occurrence across workers. @task_app.on_error handles central logging and Sentry.
• Events that must never be lost go through a transactional outbox. Use BackgroundTasks only for trivial, losable work. → background-tasks.md

Rule 22 — Async Rules

• In async def, every I/O call is awaited. Never use requests, time.sleep, sync file I/O, or CPU-heavy work (hashing, parsing) inline. Use asyncio.to_thread, or write the endpoint as a plain def if a library is sync-only.
• One shared httpx.AsyncClient from the lifespan. Every outbound call has a timeout. Retry only idempotent requests on transient errors (tenacity, backoff + jitter).
• asyncio.gather/TaskGroup for independent awaits, but never share one AsyncSession across concurrent tasks.
• Never fire-and-forget with asyncio.create_task in a request. Use an ArdiQ task.

Rule 23 — Idempotency and Webhooks

• POSTs that move money or create irreversible things accept an Idempotency-Key. Store key + request hash + response in the same transaction. The same key with the same body returns the stored response; with a different body it returns 409.
• Incoming webhooks: verify the signature over the raw body first, dedupe by event ID (unique constraint), enqueue, and return 200 in under 5 s. Re-fetch state from the provider inside the task.
• Outgoing webhooks: sent from tasks, HMAC-signed with a timestamp, with a unique event ID, retried with backoff. → api-patterns.md

Rule 24 — Streaming: Native SSE and JSON Lines

Use FastAPI's native streaming (≥ 0.135): the endpoint yields and FastAPI encodes. Never build data: ...\n\n by hand or return a StreamingResponse for SSE.

from fastapi.sse import EventSourceResponse, ServerSentEvent


@router.post(
    "",
    response_class=EventSourceResponse,
    dependencies=[Depends(rate_limit("chat", limit=20, window_s=60))],
)
async def chat(
    data: ChatRequest, user: CurrentUser, db: DbSession, llm: LlmClient
) -> AsyncIterable[ServerSentEvent]:
    # the service yields domain objects; the router wraps them in SSE events
    async for chunk in chat_service.stream_answer(db, llm, data, user):
        yield ServerSentEvent(data=chunk, event=chunk.type)

• FastAPI sets Cache-Control: no-cache and X-Accel-Buffering: no, sends keep-alive pings, and keeps dependencies open until the stream ends.
• Do the reads first, then commit() to release the pool connection during long streams.
• Errors after the first byte are sent as events (the status is already 200).
• Client disconnects cancel the generator. Cleanup writes need anyio.CancelScope(shield=True).
• JSON Lines: same yield pattern with return type AsyncIterable[Model] and no response_class. → api-patterns.md

Rule 25 — Testing

• Real Postgres (testcontainers), never SQLite. Each test runs in a rolled-back transaction (join_transaction_mode="create_savepoint"). Redis also comes from testcontainers.
• Use httpx.AsyncClient(transport=ASGITransport(app=app)) + app.dependency_overrides. Never call real third-party APIs.
• For every endpoint: happy path, 401, 403 for the wrong role, 404 for another org's resource, 422 for invalid input and unknown fields, 409 for conflicts, and side effects.
• Architecture tests: services don't import FastAPI, and every /api/ route has auth unless it's listed as public.
• Tasks: assert the enqueue via task_app.status(task_id), test the body by calling the task inline, and run each task twice to prove idempotency. → testing.md

Rule 26 — Health, Observability, Deployment

• Two health endpoints outside /api/v1: /health/live (no dependencies) and /health/ready (DB + Redis, 503 on failure).
• Sentry with environment, release (git SHA), send_default_pii=False, and a sample rate of 0.1 or less. OpenTelemetry once you have a tracing backend.
• Multi-stage Docker with uv, running as non-root. The same image serves API, worker, and migrations.
• Start the API with fastapi run --forwarded-allow-ips=<proxy>. Run workers with ardiq run app.worker:task_app. Migrations run as a one-off job before rollout. → deployment.md

Rule 27 — Naming Conventions

Rule 28 — No Over-Engineering

Don't use: a repository pattern on top of the ORM, abstract service base classes, generic CRUD base classes, event buses, DI containers, microservices before you need them, or a cache before you've measured. Use: plain async functions, direct SQLAlchemy calls, and Annotated dependencies. Duplicate twice; extract on the third time.

FastAPI Gotchas

• Route order matters: declare /tickets/export before /tickets/{ticket_id}, or "export" is parsed as an ID (→ 422).
• Types in endpoint/dependency signatures must exist at runtime. Never import them under if TYPE_CHECKING:, because FastAPI resolves annotations when it builds the route.
• List query params need Query(): tags: Annotated[list[str], Query()]. Without it, the list is read from the body.
• Dependencies are cached per request. The same dependency used in several places runs once. Use Depends(fn, use_cache=False) when you need a fresh value.
• Returning a Response with its own background= silently replaces the tasks added through the BackgroundTasks parameter.
• 204 endpoints return None. A body on a 204 is a protocol error.
• Trailing slashes: /tickets/ and /tickets are different routes, and FastAPI answers the mismatch with a 307 redirect. Declare routes without a trailing slash ("" inside a prefixed router) and have clients call exactly that.
• slowapi decorators don't work on yield endpoints. Use a dependency (Rule 18).

Security Checklist — Before Every Endpoint

• Auth dependency present (or route consciously listed as public); role in the signature/router if privileged
• Every query scoped by organization_id; other org's resources → 404
• Input schema forbids unknown fields and has no server-owned fields; strings have max_length
• Response schema excludes sensitive fields; list endpoints have page limits
• Rate limit if public, auth-related, expensive, or sends email
• Errors are AppError subclasses with no internals; no secrets or PII in logs
• Uploads validated by magic bytes + extension + size; user-supplied URLs checked for SSRF
• Tasks enqueued after commit, idempotent, with a timeout
• Tests: 401, 403, 404 cross-tenant, 422

Review Checklist — Anti-patterns to Flag