FastAPI Production Architecture

Cursor-Regel für produktionsreife Architektur- und Strukturvorgaben bei FastAPI-Services.

Original-Beschreibung der Autoren: Cursor rules for FastAPI services with router/service/repository boundaries, typed provider adapters, bulkhead isolation, idempotency, and domain exceptions.

Die Regel

---
description: "Cursor rules for FastAPI services with router/service/repository boundaries, typed provider adapters, bulkhead isolation, idempotency, and domain exceptions."
globs: **/*
alwaysApply: false
---
# FastAPI Production Architecture Rules
# Principles for production-ready FastAPI services.

## LAYER ARCHITECTURE (Principles A1-A8)

This codebase follows strict 4-layer architecture: Router → Service → Repository → ORM/HTTP/Storage.
Imports flow downward only. Each layer has hard boundaries you must NOT cross.

### Router rules (app/routers/**)
- Handlers are THIN: ≤10 lines of executable code per handler
- Allowed imports: fastapi, app.schemas.*, app.core.deps, app.services.*
- FORBIDDEN imports: sqlalchemy, httpx, boto3, app.models.*, app.repositories.*
- Every endpoint declares response_model= for OpenAPI fidelity
- Every protected/business endpoint requires user_id: str = Depends(get_current_user_id)
- Public endpoints (health checks, webhooks, callbacks) are exempt from auth
- Business logic lives in services. Routers parse input, call one service method, return response.

GOOD:
@router.post("/wallet/charge", response_model=WalletResponse, status_code=201)
async def charge(
    req: ChargeRequest,
    user_id: str = Depends(get_current_user_id),
    svc: WalletUserService = Depends(get_wallet_service),
) -> WalletResponse:
    wallet = await svc.charge(
        user_id=user_id,
        amount=req.amount,
        idempotency_key=req.idempotency_key,
    )
    return WalletResponse.from_domain(wallet)

BAD (business logic + SQL in router):
@router.post("/wallet/charge")
async def charge(req: ChargeRequest, db: Session = Depends(get_db)):
    wallet = db.query(Wallet).filter(Wallet.user_id == user_id).with_for_update().one()
    ...

### Service rules (app/services/**)
- FORBIDDEN imports: sqlalchemy, httpx, boto3, redis, FastAPI Request/Response/HTTPException
- Constructor injects Protocol-typed dependencies, not concrete classes
- Raise domain exceptions (InsufficientFundsError), not HTTPException

GOOD:
from app.repositories.protocols import WalletRepoProtocol
class WalletUserService:
    def __init__(self, repo: WalletRepoProtocol):  # Protocol, not SQLAlchemy Session
        self._repo = repo

BAD:
from sqlalchemy.orm import Session
class WalletUserService:
    def __init__(self, db: Session): ...  # Wrong — service depends on infrastructure

### Repository rules (app/repositories/**)
- ONLY layer allowed to import sqlalchemy
- Implements Protocol from app/repositories/protocols.py
- Returns domain objects, not ORM models
- Every query scoped by user_id (multi-tenancy)

### Provider rules (app/providers/**)
- ONLY layer allowed to import httpx directly
- Returns GenerateResult | ProviderError — NEVER raw dict
- Uses per-provider httpx.AsyncClient (bulkhead pattern)

## FILE SIZE RULES (Principle A1)

| LOC      | State  | Action                                      |
|----------|--------|---------------------------------------------|
| 0–399    | Green  | None.                                       |
| 400–599  | Yellow | Plan split. Add TODO(decompose) header.     |
| 600+     | Red    | BLOCK merge. Decompose first.               |

Convert file to package when ANY is true:
- Crosses 400 LOC and next change pushes past 500
- Contains 2+ disjoint sub-domains (image vs video, user vs admin)
- Mixes HTTP handlers with worker handlers
- Has 2+ callers each importing only one symbol

Safe split pattern (atomic PR):
1. Create <file>/__init__.py (empty for now)
2. Move pieces to sub-files (a.py, b.py, c.py)
3. Re-export old public names from __init__.py
4. Run tests — must pass without changes
5. Follow-up PR to migrate callers off legacy alias

__init__.py pattern:
from .user import WalletUserService
from .admin import WalletAdminService
WalletService = WalletUserService  # backwards-compat alias
__all__ = ["WalletUserService", "WalletAdminService", "WalletService"]

## EXTERNAL INTEGRATION RULES (Principles B1-B10)

### Rule 1: Anti-Corruption Layer (ACL)
Providers return GenerateResult | ProviderError, never dict.

from dataclasses import dataclass
from decimal import Decimal

@dataclass(frozen=True)
class GenerateResult:
    url: str
    cost_usd: Decimal
    latency_ms: int
    provider_request_id: str

class ProviderError(Exception):
    def __init__(self, message: str, *, retryable: bool, code: str | None = None):
        super().__init__(message); self.retryable = retryable; self.code = code

class ProviderTimeout(ProviderError):
    def __init__(self, message: str): super().__init__(message, retryable=True, code="timeout")

### Rule 2: Per-Provider Bulkhead
Each external provider has its OWN httpx.AsyncClient with its OWN Limits. NEVER share.

GOOD:
FAL_HTTP = httpx.AsyncClient(
    base_url=settings.FAL_BASE_URL,
    timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),
    limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
)
OPENAI_HTTP = httpx.AsyncClient(
    base_url="https://api.openai.com/v1",
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
)

BAD:
HTTP = httpx.AsyncClient()  # shared across all providers — no bulkhead isolation

# Shutdown cleanup — close all provider clients in FastAPI lifespan
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    yield  # app startup
    await FAL_HTTP.aclose()
    await OPENAI_HTTP.aclose()

app = FastAPI(lifespan=lifespan)
# Or on_event("shutdown"):
# @app.on_event("shutdown")
# async def close_http_clients() -> None:
#     await FAL_HTTP.aclose()
#     await OPENAI_HTTP.aclose()

### Rule 3: Idempotency Keys
Every side-effect operation accepts an idempotency_key: UUID. Look up before retrying.

### Rule 4: Structured Logging with contextvars
Use ContextVar to thread provider, user_id, request_id through async call stacks.

import contextvars

provider_var   = contextvars.ContextVar[str | None]("provider", default=None)

… (hier gekürzt — Kopieren/Download liefert die vollständige Regel)

So nutzt du sie

Die Regel kopieren (Button oben) oder als Datei herunterladen und im Projekt unter .cursor/rules/ ablegen — Cursor lädt sie beim nächsten Start automatisch. Ältere Cursor-Versionen lesen alternativ eine einzelne .cursorrules-Datei im Projektstamm; dort einfach den Regel-Text ohne den Kopfblock zwischen den ----Zeilen einfügen.

Der Regel-Text ist englisch — Cursor versteht ihn unabhängig von der Sprache, in der Sie mit dem Editor chatten.

Im Detail

Ergänzt eine reine FastAPI-Stilregel um Vorgaben für Produktionsarchitektur — Schichtentrennung (Router, Service, Repository), Konfigurationsmanagement, Logging und Fehlerbehandlung, die über einzelne Endpunkte hinausgehen. Gedacht für Teams, die eine FastAPI-Anwendung nicht als Prototyp, sondern als produktionsreifen Service aufbauen und dabei von Cursor konsistente, skalierbare Strukturvorschläge statt Ad-hoc-Code wollen. Sinnvoll in Kombination mit einer einfacheren FastAPI-Regel für den täglichen Code-Stil; diese Regel liefert eher den Rahmen für Projektstruktur und Betriebsreife.

Praxis-Tipp

Zu Beginn eines neuen Projekts einsetzen, bevor viel Code entstanden ist — nachträgliches Umstrukturieren einer bestehenden Codebasis nach der vorgeschlagenen Architektur ist aufwendiger als der gleich richtige Zuschnitt.

Lizenz & Quelle

• Lizenz: CC0 1.0
• Quelle: PatrickJS/awesome-cursorrules (GitHub)

Lade …