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 userid: str = Depends(getcurrentuserid)
• 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", responsemodel=WalletResponse, statuscode=201)

async def charge(

req: ChargeRequest,

userid: str = Depends(getcurrentuserid),

svc: WalletUserService = Depends(getwalletservice),

) -> WalletResponse:

wallet = await svc.charge(

userid=userid,

amount=req.amount,

idempotencykey=req.idempotencykey,

)

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.userid == userid).withforupdate().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)

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

providerrequestid: 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(

baseurl=settings.FALBASE_URL,

timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0),

limits=httpx.Limits(maxconnections=20, maxkeepalive_connections=10),

)

OPENAI_HTTP = httpx.Asyn