# FastAPI app structure

Worked code for the settings, app factory, per-request session, schemas, dependencies and error handlers summarised in SKILL.md.

## Contents

- Settings
- App factory and lifespan
- Database session per request
- Schemas: separate in from out
- Dependencies
- Errors

## Settings

```python
# app/settings.py
from functools import lru_cache
from pydantic import PostgresDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_", extra="ignore")

    environment: str = "development"
    database_url: PostgresDsn
    jwt_audience: str = "notes-api"
    signing_key: SecretStr
    db_pool_size: int = 10

@lru_cache
def get_settings() -> Settings:
    return Settings()
```

`SecretStr` keeps secrets out of `repr` and logs. Because `get_settings` is a
dependency, tests override it instead of patching environment variables.

## App factory and lifespan

Create shared resources once per process in the lifespan and close them on
shutdown. Module-level engines and clients make tests and workers awkward.

```python
# app/main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.settings import get_settings
from app.errors import register_error_handlers
from app.notes.router import router as notes_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    engine = create_async_engine(
        str(settings.database_url), pool_size=settings.db_pool_size, pool_pre_ping=True
    )
    app.state.sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(5.0))
    try:
        yield
    finally:
        await app.state.http.aclose()
        await engine.dispose()

def create_app() -> FastAPI:
    app = FastAPI(title="Notes API", lifespan=lifespan)
    register_error_handlers(app)
    app.include_router(notes_router, prefix="/notes", tags=["notes"])
    return app

app = create_app()
```

The old `@app.on_event("startup")` hooks are deprecated; use `lifespan`.

## Database session per request

```python
# app/db.py
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    async with request.app.state.sessionmaker() as session:
        async with session.begin():
            yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]
```

- `session.begin()` commits when the request succeeds and rolls back when any
  exception escapes, so handlers never call `commit()` themselves.
- `expire_on_commit=False` lets you return ORM objects after commit without
  triggering lazy loads, which fail in async code.
- Lazy loading of relationships raises in async; load what you need explicitly
  with `selectinload` or `joinedload`, or declare `lazy="raise"` to catch it in
  tests.

Models with SQLAlchemy 2.0 typing:

```python
# app/notes/models.py
from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(primary_key=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str]
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
```

Migrations: Alembic with the async template (`alembic init -t async`), run as a
deploy step, never from the app's lifespan.

## Schemas: separate in from out

```python
# app/notes/schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field

class NoteCreate(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    title: str = Field(min_length=1, max_length=200)
    body: str = Field(max_length=20_000)

class NoteUpdate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    title: str | None = Field(default=None, min_length=1, max_length=200)
    body: str | None = Field(default=None, max_length=20_000)

class NoteRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    body: str
    created_at: datetime
```

- Input models forbid unknown fields and never include `id` or `owner_id`.
- Output models are what the client may see; set `response_model` (or the return
  annotation) so FastAPI filters the ORM object through it.
- For PATCH, apply only fields the client sent: `data.model_dump(exclude_unset=True)`.

## Dependencies

Dependencies are FastAPI's composition tool: authentication, sessions, settings,
pagination, feature flags.

```python
# app/notes/router.py
from typing import Annotated
from fastapi import APIRouter, Depends, Query, status

from app.db import SessionDep
from app.auth import CurrentUser
from app.notes import service
from app.notes.schemas import NoteCreate, NoteRead

router = APIRouter()

class Page:
    def __init__(self, limit: Annotated[int, Query(ge=1, le=100)] = 20, after: int | None = None):
        self.limit, self.after = limit, after

@router.get("", response_model=list[NoteRead])
async def list_notes(user: CurrentUser, session: SessionDep, page: Annotated[Page, Depends()]):
    return await service.list_for_owner(session, user.id, page.limit, page.after)

@router.post("", response_model=NoteRead, status_code=status.HTTP_201_CREATED)
async def create_note(data: NoteCreate, user: CurrentUser, session: SessionDep):
    return await service.create(session, owner_id=user.id, data=data)
```

- Declare reusable aliases with `Annotated[..., Depends(...)]` (`SessionDep`,
  `CurrentUser`) instead of repeating `Depends` in every signature.
- Dependencies are cached per request: two dependencies that both need the
  session get the same one.
- Dependencies with `yield` run their cleanup after the response is produced; use
  them for anything that must be released.
- Router-level `dependencies=[Depends(require_scope("notes:write"))]` applies
  authorization to a whole group of routes.

Scope every query by the current user (`WHERE owner_id = :user_id`); see
`api-security-best-practices` for object-level authorization.

## Errors

Raise domain exceptions from services and translate them once.

```python
# app/errors.py
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

class DomainError(Exception):
    status = 400
    code = "bad-request"
    def __init__(self, detail: str):
        self.detail = detail

class NotFound(DomainError):
    status, code = 404, "not-found"

def _problem(status: int, code: str, title: str, request: Request, **extra) -> JSONResponse:
    body = {"type": f"https://api.example.com/problems/{code}", "title": title,
            "status": status, "instance": str(request.url.path), **extra}
    return JSONResponse(body, status_code=status, media_type="application/problem+json")

def register_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(DomainError)
    async def domain(request: Request, exc: DomainError):
        return _problem(exc.status, exc.code, exc.detail, request)

    @app.exception_handler(RequestValidationError)
    async def validation(request: Request, exc: RequestValidationError):
        errors = [{"loc": e["loc"], "detail": e["msg"]} for e in exc.errors()]
        return _problem(422, "validation", "Request is invalid", request, errors=errors)
```

Services then stay free of HTTP: `raise NotFound("note not found")`.