Fastapi Production Architecture Agent Rule

{
    "system_prompt": "# FastAPI Production Architecture Rules\n# Principles for production-ready FastAPI services.\n\n## LAYER ARCHITECTURE (Principles A1-A8)\n\nThis codebase follows strict 4-layer architecture: Router → Service → Repository → ORM/HTTP/Storage.\nImports flow downward only. Each layer has hard boundaries you must NOT cross.\n\n### Router rules (app/routers/**)\n- Handlers are THIN: ≤10 lines of executable code per handler\n- Allowed imports: fastapi, app.schemas.*, app.core.deps, app.services.*\n- FORBIDDEN imports: sqlalchemy, httpx, boto3, app.models.*, app.repositories.*\n- Every endpoint declares response_model= for OpenAPI fidelity\n- Every protected/business endpoint requires user_id: str = Depends(get_current_user_id)\n- Public endpoints (health checks, webhooks, callbacks) are exempt from auth\n- Business logic lives in services. Routers parse input, call one service method, return response.\n\nGOOD:\n@router.post(\"/wallet/charge\", response_model=WalletResponse, status_code=201)\nasync def charge(\n    req: ChargeRequest,\n    user_id: str = Depends(get_current_user_id),\n    svc: WalletUserService = Depends(get_wallet_service),\n) -> WalletResponse:\n    wallet = await svc.charge(\n        user_id=user_id,\n        amount=req.amount,\n        idempotency_key=req.idempotency_key,\n    )\n    return WalletResponse.from_domain(wallet)\n\nBAD (business logic + SQL in router):\n@router.post(\"/wallet/charge\")\nasync def charge(req: ChargeRequest, db: Session = Depends(get_db)):\n    wallet = db.query(Wallet).filter(Wallet.user_id == user_id).with_for_update().one()\n    ...\n\n### Service rules (app/services/**)\n- FORBIDDEN imports: sqlalchemy, httpx, boto3, redis, FastAPI Request/Response/HTTPException\n- Constructor injects Protocol-typed dependencies, not concrete classes\n- Raise domain exceptions (InsufficientFundsError), not HTTPException\n\nGOOD:\nfrom app.repositories.protocols import WalletRepoProtocol\nclass WalletUserService:\n    def __init__(self, repo: WalletRepoProtocol):  # Protocol, not SQLAlchemy Session\n        self._repo = repo\n\nBAD:\nfrom sqlalchemy.orm import Session\nclass WalletUserService:\n    def __init__(self, db: Session): ...  # Wrong — service depends on infrastructure\n\n### Repository rules (app/repositories/**)\n- ONLY layer allowed to import sqlalchemy\n- Implements Protocol from app/repositories/protocols.py\n- Returns domain objects, not ORM models\n- Every query scoped by user_id (multi-tenancy)\n\n### Provider rules (app/providers/**)\n- ONLY layer allowed to import httpx directly\n- Returns GenerateResult | ProviderError — NEVER raw dict\n- Uses per-provider httpx.AsyncClient (bulkhead pattern)\n\n## FILE SIZE RULES (Principle A1)\n\n| LOC      | State  | Action                                      |\n|----------|--------|---------------------------------------------|\n| 0–399    | Green  | None.                                       |\n| 400–599  | Yellow | Plan split. Add TODO(decompose) header.     |\n| 600+     | Red    | BLOCK merge. Decompose first.               |\n\nConvert file to package when ANY is true:\n- Crosses 400 LOC and next change pushes past 500\n- Contains 2+ disjoint sub-domains (image vs video, user vs admin)\n- Mixes HTTP handlers with worker handlers\n- Has 2+ callers each importing only one symbol\n\nSafe split pattern (atomic PR):\n1. Create <file>/__init__.py (empty for now)\n2. Move pieces to sub-files (a.py, b.py, c.py)\n3. Re-export old public names from __init__.py\n4. Run tests — must pass without changes\n5. Follow-up PR to migrate callers off legacy alias\n\n__init__.py pattern:\nfrom .user import WalletUserService\nfrom .admin import WalletAdminService\nWalletService = WalletUserService  # backwards-compat alias\n__all__ = [\"WalletUserService\", \"WalletAdminService\", \"WalletService\"]\n\n## EXTERNAL INTEGRATION RULES (Principles B1-B10)\n\n### Rule 1: Anti-Corruption Layer (ACL)\nProviders return GenerateResult | ProviderError, never dict.\n\nfrom dataclasses import dataclass\nfrom decimal import Decimal\n\n@dataclass(frozen=True)\nclass GenerateResult:\n    url: str\n    cost_usd: Decimal\n    latency_ms: int\n    provider_request_id: str\n\nclass ProviderError(Exception):\n    def __init__(self, message: str, *, retryable: bool, code: str | None = None):\n        super().__init__(message); self.retryable = retryable; self.code = code\n\nclass ProviderTimeout(ProviderError):\n    def __init__(self, message: str): super().__init__(message, retryable=True, code=\"timeout\")\n\n### Rule 2: Per-Provider Bulkhead\nEach external provider has its OWN httpx.AsyncClient with its OWN Limits. NEVER share.\n\nGOOD:\nFAL_HTTP = httpx.AsyncClient(\n    base_url=settings.FAL_BASE_URL,\n    timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),\n    limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),\n)\nOPENAI_HTTP = httpx.AsyncClient(\n    base_url=\"https://api.openai.com/v1\",\n    limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),\n)\n\nBAD:\nHTTP = httpx.AsyncClient()  # shared across all providers — no bulkhead isolation\n\n# Shutdown cleanup — close all provider clients in FastAPI lifespan\nfrom contextlib import asynccontextmanager\n\n@asynccontextmanager\nasync def lifespan(app):\n    yield  # app startup\n    await FAL_HTTP.aclose()\n    await OPENAI_HTTP.aclose()\n\napp = FastAPI(lifespan=lifespan)\n# Or on_event(\"shutdown\"):\n# @app.on_event(\"shutdown\")\n# async def close_http_clients() -> None:\n#     await FAL_HTTP.aclose()\n#     await OPENAI_HTTP.aclose()\n\n### Rule 3: Idempotency Keys\nEvery side-effect operation accepts an idempotency_key: UUID. Look up before retrying.\n\n### Rule 4: Structured Logging with contextvars\nUse ContextVar to thread provider, user_id, request_id through async call stacks.\n\nimport contextvars\n\nprovider_var   = contextvars.ContextVar[str | None](\"provider\", default=None)\nuser_id_var    = contextvars.ContextVar[str | None](\"user_id\", default=None)\nrequest_id_var = contextvars.ContextVar[str | None](\"request_id\", default=None)\n\n# Read in logs: provider_var.get(), user_id_var.get(), request_id_var.get()\n# JSON formatter picks these up automatically via extra={} or ContextVar.get()\n\n### Rule 5: Single-Writer Principle (Principle B10)\nFor safety-critical state: exactly ONE service-layer module does the writing.\nOnly the designated writer service for a domain may call repo.hold().\nRouters and providers must NOT call repo.hold() directly.\nAdmin services implemented in the service layer may call repo.hold() only\nif they are the designated writer for that domain.\nEnforce via: code-review grep check (`grep -r \"repo\\.hold(\" --include=\"*.py\"`)\nand unit tests that assert call-origin of repo.hold().\n\n## ANTI-PATTERNS — REJECT ON SIGHT\n\n1. def some_method(self, db: Session, ...) → use Protocol-typed repo\n2. from app.models.user import User inside a service → return domain types from repo\n3. httpx.AsyncClient() instantiated inside a function → use shared per-provider client\n4. raise HTTPException(...) inside a service → raise domain exception\n5. db.query(...) inside a router → move to service then repo\n6. async def call(self, req) -> dict: return resp.json() → ACL violation\n7. HTTP = httpx.AsyncClient() shared → bulkhead violation\n8. result[\"vendor_field\"][\"nested\"] in service code → ACL violation\n9. logger.info(f\"{user_id} did X\") → use structured logging with extra={}\n10. Side-effect operation without idempotency_key → double-charge risk\n\n## DEPENDENCY INJECTION\n\nUse FastAPI Depends() + factory functions in app/core/deps.py.\nDo NOT install dependency-injector, punq, or any DI container.\n\ndef get_wallet_service(db: Session = Depends(get_db)) -> WalletUserService:\n    return WalletUserService(repo=SQLAlchemyWalletRepo(db))\n\n## DOMAIN EXCEPTIONS PATTERN\n\nServices raise domain errors. Routers map to HTTP.\n\n# services/wallet/exceptions.py\nclass InsufficientFundsError(Exception): ...\nclass WalletNotFoundError(Exception): ...\n\n# routers/wallet.py\ntry:\n    wallet = await svc.charge(...)\nexcept InsufficientFundsError:\n    raise HTTPException(402, detail=\"insufficient funds\")",
    "prompt_type": "agent_rule",
    "framework": "cursor",
    "globs": "**/*",
    "compatible_models": [
        "Claude 3.5 Sonnet",
        "GPT-4o",
        "Cursor AI",
        "Gemini 2.5 Flash"
    ],
    "download_filename": "fastapi-production-architecture.cursorrules",
    "tags": [
        "cursor",
        "cursorrules",
        "agent",
        "coding",
        "fastapi"
    ]
}