Task-oriented walkthroughs for people changing the code. Written for a rotating student-org team with mixed FastAPI/Postgres experience: if a step feels obvious, skip it; if it doesn't, it's spelled out.
Read ARCHITECTURE.md first — the walkthroughs below assume you know
the three Protocol boundaries.
- Follow the Protocol boundary.
contracts/is the stable interface layer and imports nothing fromsrc/. Domain logic (src/ingest.py,src/url_norm.py) and routers depend on Protocols, never on a concrete adapter. Concrete wiring lives only insrc/api/deps.py. - If it touches persistence, it goes through
StorageAdapter— and therefore through both adapters (in-memory and Postgres), or the fast tests and prod diverge. - Terminology: "storage adapter", "Protocol boundary", "scoped API key", "source of truth", "degrade-on-directory-down", "ingest". Match the existing docs.
- Keep the fast suite fast and Docker-free. New behavior gets an in-memory test; only storage-adapter parity needs the Postgres suite.
- Lint before you push:
uv run ruff check .anduv run ruff format --check .(line length 100, target py311). CI gates both.
# From the repo root, enter the service directory first
cd services/documentation-system
uv sync --extra dev
docker compose up -d postgres # only needed for the Postgres test suite / running the serverNote: the repo is a uv workspace (one root
pyproject.toml, one rootuv.lockshared withpackages/authand team-tracking). documentation-system depends on the sharedplatform-authpackage via[tool.uv.sources] platform-auth = { workspace = true }, butuv sync/uv runfrom this directory work exactly as shown above.
Endpoints live in src/api/routers/ (docs.py, sources.py), grouped by resource.
- Pick the router (or add a new
APIRouterandinclude_routerit insrc/api/app.py). - Declare request/response models. Response models are the domain types in
contracts/types.py; request bodies are small Pydantic models withmodel_config = ConfigDict(extra="forbid")so unknown fields 422 instead of being silently dropped (seeTagBodyindocs.py). - Gate it with a scope. Add
_: AuthedKey = Depends(require_scope("docs:read"))for reads or"docs:write"for writes. If it mutates and needs an actor, also takeactor: str = Depends(get_actor)and pass it to the storage call. - Inject dependencies via
Depends:get_storage, andget_fetchers/get_directoryif you need them. Never construct adapters inline — that's what makes the route testable. - Map domain errors to HTTP.
Nonefrom storage →raise HTTPException(404, …); catchBadReference→ 400; catchFetchError→ 502. Follow the patterns already indocs.py. - Test it in
tests/test_api_docs.py(or a new file), using the fixtures that overrideget_storagewithInMemoryStorageAdapter. Assert status codes and body.
Goal: give an auth-gated source (e.g. Notion) a real content snapshot.
- Implement the
FetcherProtocol — a class withdef fetch(self, url: str) -> FetchResult. Put it insrc/fetch/<source>.py. Return aFetchResult(title=…, content_snapshot=…); raiseFetchErroron any retrieval/parse failure (never return a half-broken result). Use a timeout on network calls — seeWebFetcherfor the httpx pattern. - Register it by
source_idindefault_registry()(src/fetch/registry.py):The key must equal the source'sreturn FetcherRegistry({ "web": WebFetcher(), "github": GithubFetcher(), "notion": NotionFetcher(), # new })
id— that's howfetch_forfinds it. - Enable fetching for that source. Flip
content_fetch_enabledtotruefor the source. That's a data change in thesourcestable, so write a migration that updates the seeded row (see below). Ingest only attempts a fetch when the source'scontent_fetch_enabledis true and a fetcher is registered. - Handle auth. If the source
requires_auth, your fetcher needs credentials. There's no fetcher-specific config wired in v1 — add a setting tosrc/config.py(and document it inDEPLOYMENT.md) and thread it throughdefault_registry(). - Test in
tests/test_fetchers.py: parse-a-known-page, and theFetchErrorpath. Fetchers accept an injected httpx client, so tests pass a fake instead of hitting the network.
Any new persistence operation is a three-part change — skip a part and the adapters drift.
- Declare it on the Protocol in
contracts/storage.py, with a docstring specifying the exact semantics (return type, what "not found" returns, idempotency). This is the contract both adapters must honor. - Implement it in both adapters, identically:
src/storage/in_memory.py— operate on the dicts (self._docs,self._tags, …). Re-hydrate docs with their tags via_hydratebefore returning.src/storage/postgres.py— SQLAlchemy Core against the tables insrc/storage/schema.py. Keep the same return contract (e.g.Nonefor missing).
- Test parity. Add the behavioral assertion to the in-memory test
(
tests/test_in_memory_adapter.py) and mirror it intests/test_postgres_adapter.pyso theRUN_PG_TESTSsuite proves the two agree. Usebuild_seed_sources()fromconftest.pyfor a consistent starting state.
Rule of thumb: if the in-memory suite passes but you didn't touch postgres.py, you
probably introduced a divergence.
Schema and seed data changes go through Alembic (migrations/versions/). Migrations are
numbered sequentially (001, 002, …) and chained by down_revision.
- Create
migrations/versions/003_<short_name>.py. Setrevision = "003"anddown_revision = "002"(the current head). Copy the header/imports from an existing migration. - Write
upgrade()and a realdowngrade()— every migration must be reversible. Useop.create_table,op.add_column,op.bulk_insert,op.execute(sa.text(...)), etc. Migration002is a good template for seed-data changes. - Keep
src/storage/schema.pyin sync. The Core table definitions there must match the post-migration schema — the Postgres adapter and the migrations share that shape. - Apply and verify:
uv run alembic upgrade head uv run alembic downgrade -1 && uv run alembic upgrade head # prove downgrade works
Two modes:
Fast (default — in-memory, no Docker):
uv run pytest --ignore=tests/test_postgres_adapter.py -q
# 59 passedUses InMemoryStorageAdapter injected via dependency_overrides, plus fakes for the
Fetcher and DirectoryClient. Runs in well under a second. This is the suite you run on
every change.
Full (adds Postgres integration):
docker compose up -d postgres
RUN_PG_TESTS=1 uv run pytest -qtests/test_postgres_adapter.py runs the same behavioral assertions against a live
database and is gated behind RUN_PG_TESTS=1 — without the flag it's skipped (running the
whole suite without the flag reports 59 passed, 7 skipped). Run this before any change
that touches postgres.py, schema.py, or a migration.
uv run ruff check . # lint
uv run ruff check --fix . # auto-fix what it can
uv run ruff format . # format
uv run ruff format --check . # verify without writing (CI-style)Ruff config lives in pyproject.toml (line length 100, target py311). Keep the tree
warning-clean.