FastAPI

# CLAUDE.md — FastAPI + Pydantic + SQLAlchemy

## Project layout

```
src/myapp/
  api/
    routers/         # one file per resource
    deps.py          # shared FastAPI dependencies
  core/
    config.py        # pydantic-settings
    security.py
  db/
    base.py          # declarative_base
    session.py       # engine + sessionmaker
  models/            # SQLAlchemy ORM models
  schemas/           # Pydantic models for I/O
  services/          # business logic
  main.py            # app = FastAPI(); include_router(...)
```

- Models, schemas, and services have separate files. Don't smush them into one `models.py`.

## Pydantic v2

- Use Pydantic v2. `BaseModel` only — no v1 syntax.
- One schema per direction: `UserCreate`, `UserUpdate`, `UserRead`. Never reuse a "User" schema across all three.
- `model_config = ConfigDict(from_attributes=True)` on read schemas to read from ORM objects.
- Validate at the schema. `@field_validator` for single-field rules; `@model_validator` for cross-field.

## SQLAlchemy 2.0

- Use the `Mapped[...]` typed style. No legacy `Column(...)` declarations.
  ```python
  class User(Base):
      __tablename__ = "users"
      id: Mapped[int] = mapped_column(primary_key=True)
      email: Mapped[str] = mapped_column(unique=True)
  ```
- Use `select(...)` and the modern execute API. No `session.query(...)`.
- Sessions are scoped to a request via a FastAPI dependency:
  ```python
  def get_db() -> Iterator[Session]:
      with SessionLocal() as session:
          yield session
  ```
- Commit at the boundary (the endpoint or service), never inside a repository helper.

## Dependency injection

- FastAPI `Depends` is your DI. Use it for db sessions, auth users, settings, feature flags.
- One `deps.py` per app for cross-cutting deps. Per-router deps stay in the router.
- Don't create dependencies that have side effects (logging a metric is fine; mutating state is not).

## Routers

- One router per resource. `prefix="/users"`, `tags=["users"]`.
- Endpoints are thin: parse → call service → return. No business logic in the route function.
- `response_model=UserRead` on every endpoint that returns data. FastAPI strips fields not in the schema.

## Errors

- Raise `HTTPException` for known failure modes with the right status code.
- Custom exception classes inherit from a base; register handlers via `@app.exception_handler(...)`.
- Don't return error dicts from endpoints. Raise.

## Don't

- Don't share a Session across requests. Use `Depends(get_db)`.
- Don't run schema migrations in app startup. Use Alembic and a separate migration step.
- Don't put SQL queries in routers. Wrap them in a repository or service.
- Don't expose ORM models directly from endpoints. Always go through a `*Read` Pydantic schema.

# CLAUDE.md — FastAPI Async + Postgres

## Async stack

- Driver: **asyncpg** (fastest, async-native).
- ORM: SQLAlchemy 2.0 with `AsyncEngine` + `AsyncSession`. Or **SQLModel** if you want Pydantic-flavored models.
- Migrations: **Alembic** in async mode (`async def run_migrations_online`).

## Engine & sessions

```python
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,
    max_overflow=10,
    pool_pre_ping=True,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
```

- `expire_on_commit=False` so attributes stay accessible after commit.
- Pool sizing: start with `pool_size = workers × concurrent_requests`. Adjust under load.
- One async session per request, via a FastAPI dependency:
  ```python
  async def get_db() -> AsyncIterator[AsyncSession]:
      async with SessionLocal() as session:
          yield session
  ```

## Async patterns

- Every endpoint is `async def`. Sync defs run in a thread pool — usually wrong here.
- Every DB call: `await session.execute(stmt)`. Forgetting `await` returns a coroutine and a confusing error.
- Use `asyncio.gather` to run independent reads in parallel:
  ```python
  user, orders = await asyncio.gather(get_user(id), get_orders(id))
  ```

## Connection pooling

- For serverless / many short-lived containers, use **pgbouncer** in transaction mode in front of Postgres. Configure SQLAlchemy with `pool_pre_ping=True` and `connect_args={"prepared_statement_cache_size": 0}` for asyncpg.
- For long-lived processes (k8s deploy), pool inside the app with `pool_size`. Don't double-pool.
- `DATABASE_URL` for the pooled URL; `DIRECT_URL` for migrations (Alembic needs direct).

## Transactions

- Begin a transaction at the boundary, not per-statement:
  ```python
  async with session.begin():
      ...  # one atomic unit
  ```
- Read-only endpoints don't need an explicit transaction.
- Don't catch and swallow `IntegrityError` — let it propagate, map to `HTTPException(409)` at the boundary.

## Migrations

- Alembic with `revision --autogenerate` then **always review** the diff. Autogenerate misses constraints, indexes, and enum changes.
- Long migrations run as a separate job, not at app boot.
- Backward-compatible deploys: add nullable column → deploy code → backfill → make NOT NULL.

## Performance

- Add indexes for every WHERE/ORDER column you actually query.
- Use `selectinload` / `joinedload` to avoid N+1 — pick one per relationship and stick with it.
- For bulk inserts, use `session.execute(insert(Model), [...])`. Don't loop `session.add`.

## Don't

- Don't mix sync and async sessions. Pick one path; convert legacy code, don't bridge.
- Don't share an `AsyncSession` across coroutines. One per task.
- Don't use the default psycopg2 sync driver in an async app. It blocks.
- Don't run `await` calls in `__init__`. Use an async factory or lifespan.

# CLAUDE.md — FastAPI JWT Authentication

## Tokens

- Two tokens: **access token** (short-lived, ~15 min) + **refresh token** (longer, ~7–30 days).
- Sign with HS256 if you have one service. Switch to RS256 if you have multiple services that need to verify without sharing the signing key.
- Library: **python-jose** or **PyJWT**. Either is fine — pick one.
- Claims: `sub` (user id), `iat`, `exp`, `jti` (for revocation), and a `type` claim distinguishing access vs refresh.

## Storage on the client

- Web: HTTP-only, Secure, SameSite=Lax cookies for the refresh token. Access token in memory only.
- Mobile / native: secure storage (Keychain, EncryptedSharedPreferences). Never plain disk.
- Don't put tokens in `localStorage`. XSS reads it.

## Password hashing

- **bcrypt** via `passlib` or **argon2** via `argon2-cffi`. Argon2 if starting fresh.
- Cost factor: bcrypt 12+, argon2 default. Re-evaluate yearly.
- Never log passwords. Filter them out of error reporting.

## FastAPI integration

- Auth dependency:
  ```python
  async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> User:
      try:
          payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
      except JWTError:
          raise HTTPException(401, "Invalid token")
      user = await db.get(User, payload["sub"])
      if not user:
          raise HTTPException(401, "Invalid token")
      return user
  ```
- For optional auth, write a separate dependency that returns `Optional[User]`. Don't make `get_current_user` accept `None`.

## Refresh flow

- Refresh endpoint: takes the refresh token, issues new access + refresh, **rotates** the refresh token.
- Track refresh tokens in a `refresh_tokens` table with `jti`, `user_id`, `expires_at`, `revoked_at`. Lookup on each refresh.
- On suspicious activity (token reuse), revoke the entire user session family.

## Roles & permissions

- Role on the user record, claim copied to the access token. Don't re-fetch the user just for role checks.
- Per-endpoint authorization: factory dependencies like `require_role("admin")`.
- For object-level permissions, check ownership inside the service — JWT only proves identity, not authorization.

## Logout

- Stateless JWTs can't truly be invalidated. For real logout:
  - Delete the refresh-token row (the user can't refresh again)
  - Optionally maintain a revocation list keyed by `jti` for unexpired access tokens
  - Or shorten access-token TTL so revocation is "soon enough"

## Don't

- Don't use long-lived access tokens to avoid refresh logic. The refresh flow exists for revocation.
- Don't put sensitive data (email, role, plan) you'd hate to leak in the JWT payload — anyone can read it.
- Don't trust `iss` and `aud` without configuring them. Set them in encode and verify in decode.
- Don't roll your own crypto. Use library-supported algorithms only.

# CLAUDE.md — FastAPI Testing (pytest + httpx)

## Stack

- **pytest** + **pytest-asyncio** + **httpx**.
- Use `httpx.AsyncClient` with the FastAPI app as transport — not the legacy `TestClient` synchronously, unless you have purely sync code.
- `anyio` is fine too if you prefer the trio-compatible style.

## Async client fixture

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

- Use `ASGITransport` so the request never touches a real socket.
- Keep `base_url` consistent so logs are readable.

## Database isolation

- Spin up a Postgres container for tests (Docker Compose, Testcontainers, or a local `pg_tmp`). SQLite isn't equivalent.
- Each test runs in a transaction that rolls back at the end:
  ```python
  @pytest.fixture
  async def db(engine):
      async with engine.connect() as conn:
          trans = await conn.begin()
          session = AsyncSession(bind=conn, expire_on_commit=False)
          yield session
          await session.close()
          await trans.rollback()
  ```
- Override `get_db` via `app.dependency_overrides` so the endpoint uses the test session.

## Factories

- **factory_boy** with async-friendly post-generation hooks, or just plain async factory functions.
- Factories return persisted objects unless you ask for `build`-only.
- Don't share factory state across tests — every test creates what it needs.

## Auth in tests

- Override `get_current_user` to return a test user. Don't hit the real JWT issue endpoint in every test.
- Have a separate test category that *does* exercise the real auth flow — keep it small and slow.

## What to test

- **Endpoint contract**: status codes, response shape, validation errors. Use Pydantic schemas to assert on shape.
- **Side effects**: row was created, email task was enqueued, cache was invalidated.
- **Edge cases**: missing fields, unauthorized users, conflicting state.

## Speed

- `--reuse-db` equivalent: reuse the test DB across runs by skipping migrations on every test.
- `pytest -n auto` with `pytest-xdist` parallelism. Make sure factories don't collide on unique constraints.
- Profile with `pytest --durations=10`. Fix the worst offenders before they multiply.

## Don't

- Don't make HTTP calls to a running dev server in tests. Use the app directly via ASGI.
- Don't mock `httpx.AsyncClient` — test against the real client with a stubbed transport when calling external services.
- Don't rely on insertion order. Use deterministic ordering in queries when asserting.
- Don't share authenticated user state across tests via session middleware.

# CLAUDE.md — FastAPI Production Deploy

## Image

- Two-stage Dockerfile. Builder installs deps, runtime copies the `/app` and the venv.
- Base: `python:3.12-slim-bookworm`. Pin minor version.
- Non-root user: `RUN adduser --disabled-password app && USER app`.
- One process per container. The image runs `uvicorn`; orchestration handles replicas.

## ASGI server

- **uvicorn** behind a real ASGI gateway (`uvicorn[standard]` for HTTP/2 and websockets):
  ```sh
  uvicorn myapp.main:app --host 0.0.0.0 --port 8000 --workers 1
  ```
- Use **gunicorn with uvicorn workers** when you want process-level concurrency: `gunicorn -k uvicorn.workers.UvicornWorker -w 4`.
- For pure CPU-bound work, you don't want async — split into a worker queue.
- `--workers` count: `2 × CPU` is a starting point. Profile.

## Lifespan & startup

- Use the modern `lifespan` context manager, not `@app.on_event("startup")`:
  ```python
  @asynccontextmanager
  async def lifespan(app: FastAPI):
      await connect_db()
      yield
      await disconnect_db()
  ```
- Open connections in `lifespan`, not at import time. Otherwise multi-worker setups create N pools.

## Health checks

- `/health/live` — process is alive. Returns 200 unconditionally.
- `/health/ready` — process can serve traffic (DB reachable, cache reachable, migrations applied). Returns 503 on failure.
- Don't gate liveness on the DB. A dead DB shouldn't kill the pod that's mid-graceful-shutdown.

## Kubernetes

- Resource requests and limits set on every deployment. CPU `request` ≈ steady-state; `limit` 2× request or unset.
- HPA on CPU + custom metrics (request rate). Pure CPU isn't enough for async workloads.
- Probes:
  - `livenessProbe`: `/health/live`, period 10s
  - `readinessProbe`: `/health/ready`, period 5s
  - `startupProbe` for slow boots, period 5s, `failureThreshold` 30
- `terminationGracePeriodSeconds` ≥ uvicorn's graceful shutdown timeout.

## Observability

- **OpenTelemetry** instrumentation: `opentelemetry-instrumentation-fastapi`. Auto-traces every request.
- Logs: JSON to stdout, fields for `trace_id`, `span_id`, `user_id` where available.
- Metrics: Prometheus exposed at `/metrics` (use `prometheus-fastapi-instrumentator`).
- Sentry for exceptions, gated on `SENTRY_DSN` env var.

## Configuration

- `pydantic-settings` reads env into a typed `Settings` object. Single instance, imported once.
- 12-factor: every config from env, no config files in the image.
- Secrets injected by the orchestrator (Kubernetes Secrets, AWS Secrets Manager, etc.). Never baked.

## Don't

- Don't run uvicorn in `--reload` mode in production.
- Don't put migrations in `lifespan`. Run them as a one-shot Job before rolling out the new ReplicaSet.
- Don't ship without limits — a runaway worker can exhaust the node.
- Don't rely on `sys.exit` for shutdown. Cleanly close connections and let uvicorn drain.