# Writing a source adapter An **adapter** turns one publisher's data into typed records. It never touches HTTP or the database directly: the framework fetches (politely, legally, archived) and the Loader writes. Your job is two functions, `start()` and `parse()`, and a description of the publisher (`SourceSpec`). ``` start(ctx) -> Request ... entry points (list pages, feeds, API queries) │ Fetcher: robots.txt, throttle, TDM check, conditional GET, retries, raw archive ▼ parse(ctx, resp) -> Record | Request ... PURE: no DB, no network, no clock-dependent output │ Loader: entity resolution, claim identity, evidence, facets, corroboration status ▼ PostgreSQL ``` Read [architecture/01-design-from-scratch.md](../architecture/01-design-from-scratch.md) §4 for the reasoning; this page is the how-to. ## 0. Before writing code: the legal checklist Answer these in the pull request (they end up in `SourceSpec.tos_notes`): 1. **Is there an official API, bulk download or feed?** Use it instead of scraping HTML. 2. **robots.txt** – the Fetcher enforces it. Never work around it. If a documented public API is listed in robots.txt only because of generic crawler rules (Wikidata SPARQL, the MediaWiki API), add the API prefix to `robots_exempt` and cite the operator's API policy in `tos_notes`. 3. **Terms of service / licence** – may automated access happen? May we keep full text? If not, `store_fulltext=False` (only title, short excerpt and link are kept) and usually `keep_raw=False`. 4. **TDM reservation** (§42h UrhG, W3C TDMRep) – detected automatically per origin. If set, no text and no raw archive are kept, whatever the spec says. 5. **Personal data** – only data about people in their public role. There is no address field in any record type; don't smuggle one into `data`. 6. **No circumvention** – no paywall tricks, no mirrors/proxies of platforms whose ToS forbid scraping (e.g. Nitter for X), no logged-in sessions, no private groups. If any answer is "no" or "unclear", document the source in [docs/sources.md](../sources.md) under "Approaches we deliberately do not use" instead. ## 1. Where things go ``` backend/politrack/sources/ my_source.py one module per publisher (or a package for large ones, see parlament/) backend/tests/fixtures// .request.json captured request (url, kind, meta) .html|json|xml|bin captured response body .expected.json golden output of parse() ``` Modules under `politrack/sources` and `politrack/enrich` are imported automatically; `@register` is all it takes. External packages can add adapters through the `politrack.adapters` entry-point group (see `backend/pyproject.toml`). ## 2. A complete example A (fictitious) state parliament publishes its members as JSON and a page per member: ```python """Landtag Musterland: members (JSON API) and their declared side jobs (HTML).""" from __future__ import annotations from collections.abc import Iterator from typing import Any from politrack.core import Adapter, Request, Response, RunContext, SourceSpec, register from politrack.core.records import Claim, Evidence, OrgRef, PersonRecord, PersonRef, Role from politrack.text import excerpt BASE = "https://www.landtag-musterland.at" SPEC = SourceSpec( slug="landtag-musterland", # stable; referenced by Evidence.source and user filters name="Landtag Musterland", kind="official", # official | ngo | media | social | aggregator | user | derived homepage=BASE, license="Amtliches Werk (§7 UrhG)", access="api", # api | bulk | rss | scrape | manual | derived trust=90, # default trust 0-100 (users can override) min_interval=2.0, # seconds between requests to this host tos_notes="Open JSON API, no terms restricting reuse; robots.txt allows /api/.", ) @register class MusterlandMembers(Adapter): spec = SPEC name = "musterland.members" # .; also the CLI/worker task name description = "Members of the Landtag Musterland with mandates and side jobs" schedule = "15 4 * * *" # cron (UTC); None = manual only queue = "scrape" # scrape | social | bulk creates_persons = True # authoritative member list: may create persons parser_version = 1 # bump when parse() output changes, then `politrack reparse` def start(self, ctx: RunContext) -> Iterator[Request]: yield Request(url=f"{BASE}/api/members", params={"period": ctx.params.get("period", "current")}, kind="list") def parse(self, ctx: RunContext, response: Response) -> Iterator[Any]: if response.request.kind == "list": for m in response.json()["members"]: ref = PersonRef(ids={"landtag-musterland": str(m["id"])}, name=m["name"]) yield PersonRecord(ref=ref, full_name=m["name"], photo_url=m.get("photo")) yield Claim( claim_type="role", person=ref, summary_de=f"{m['name']} ist Abgeordnete:r zum Landtag Musterland ({m['party']})", summary_en=f"{m['name']} is a member of the Musterland state parliament ({m['party']})", valid_from=m.get("since"), evidence=[Evidence(url=f"{BASE}/abgeordnete/{m['id']}", doc_type="api_record", title=m["name"])], facet=Role(role="mp", level="state", region="Musterland", body=OrgRef(slug="landtag-musterland", name="Landtag Musterland", kind="parliament"), party=OrgRef(name=m["party"], kind="party", create=False), start=m.get("since")), ) # follow-up request; meta travels with it to parse() yield Request(url=f"{BASE}/abgeordnete/{m['id']}", kind="profile", meta={"member_id": str(m["id"]), "name": m["name"]}) return # kind == "profile": HTML page with a side-job table for row in response.soup.select("table.nebentaetigkeiten tr")[1:]: employer, function = (td.get_text(" ", strip=True) for td in row.find_all("td")[:2]) yield Claim( claim_type="side_job", person=PersonRef(ids={"landtag-musterland": response.meta["member_id"]}), counterpart_org=OrgRef(name=employer, kind="company"), key=("musterland", response.meta["member_id"], employer, function), # no facet -> explicit key summary_de=f"{response.meta['name']}: {function} bei {employer}", evidence=[Evidence(url=response.url, doc_type="register_entry", quote=excerpt(row.get_text(" "), 300))], ) ``` Run it: ```bash docker compose exec api politrack sources sync # registers SPEC in the DB docker compose exec api politrack run musterland.members --max-requests 20 -p period=current docker compose exec api politrack status # run stats, errors ``` The periodic task appears automatically for the workers (`schedule`, `queue`). ## 3. The building blocks ### Request | field | purpose | |---|---| | `url`, `method`, `params`, `json`, `headers` | the HTTP request. Secrets go in `params` named `key`/`token`/`api_key`/`access_token`, or in `headers` – both are redacted from the archive and logs. | | `kind` | dispatch key for `parse()` | | `meta` | context for `parse()` (ids, names, depth); not part of the request identity | | `conditional` | send `If-None-Match`/`If-Modified-Since` from the archive (default on) | | `skip_unchanged` | when body hash and `parser_version` are unchanged, don't reload records – only follow-up requests are collected, so an interrupted run resumes (default on) | A request's identity (`fingerprint`) is method + url + params + json. The runner de-duplicates within a run. ### Response `response.text`, `response.json()`, `response.soup` (BeautifulSoup, `html.parser`), `response.meta` (= `request.meta`), `response.url` (final URL, secrets redacted), `response.tdm_reserved`. ### RunContext `ctx.params` (CLI `-p key=value`, worker parameters), `ctx.state` (persistent cursors, see below), `ctx.stats` (a Counter shown in `politrack status` – count what you skip and why), `ctx.log`. `ctx.conn` is available in `start()` only (e.g. to read which social accounts to poll); **`parse()` must not query the database** – that is what keeps it replayable and testable. ### Records All in `politrack/core/records.py`: | record | use for | |---|---| | `PersonRecord`, `OrgRecord` | entity data from authoritative sources (names, photo, dates) | | `SocialAccount` | an account of a person/organisation (`verified=True` only if an official page links it) | | `VoteEvent`, `Event` | things with a key that claims point at (votes, sessions, trips) | | `Claim` (+ `facet`) | every fact. Facets: `Role`, `VotePosition`, `Money`, `Participation`, `Statement`, `Legal` | | `Document` | a document without a claim yet (headlines, press releases) – enrichers derive claims later | Claim types: `role vote donation money_flow statement appearance meeting travel side_job legal_proceeding lobbying media_mention career parliamentary_question government_decision wiki_edit declaration ownership affiliation party_finance other` (see [data-model.md](../data-model.md)). ### Referring to people and organisations Never look up ids yourself. Describe what you know and let the Loader resolve it: ```python PersonRef(ids={"parlament": "2339"}) # strong identifier (best) PersonRef(ids={"wikidata": "Q78869"}, name="Alexander Van der Bellen") PersonRef(name="Leonore Gewessler") # exact alias match (titles stripped) PersonRef(hints={"surname_in_body": "nationalrat", "on": date(2024, 3, 1)}, name="Kickl") # roll calls PersonRef(hints={"office": "Bundesministerin für Justiz", "on": date(2025, 6, 17)}) # office holder OrgRef(slug="oevp") # seeded organisations OrgRef(ids={"firmenbuch": "123456a"}, name="Muster GmbH", kind="company") # created if unknown OrgRef(name="ÖVP", kind="party", create=False) # match only, never create ``` Identifier schemes: `parlament`, `wikidata`, `ep`, `uid`, `firmenbuch`, `rtr`, `eu_transparency`, `google_advertiser`, `facebook_page` – add a new scheme by simply using it (document it in data-model.md). Persons are only created by adapters with `creates_persons = True` (authoritative member lists); everyone else's claims about unknown people are skipped and counted as `unresolved_person`. Organisation names are matched case-, accent- and legal-form-insensitively (`normalize_org_name`), so "HYPO TIROL BANK AG" and "Hypo Tirol Bank Aktiengesellschaft" become one organisation. ### Claim identity – facet or `key` The Loader computes a claim's identity from *resolved* entities and the facet: a donation is (kind, payer, payee, amount, date, period, reference), a vote position is (vote, person/party), a role is (person, body, role, start), a statement is (person/org, url), a participation is (event, person/org), a media mention is (url, person/org). So the same fact reported by two sources becomes **one claim with two pieces of evidence** – that is how corroboration works. Claims **without a facet** (side jobs, careers, declarations, holdings, questions …) need an explicit `key` – the Loader refuses them otherwise. Build it from what identifies the fact in the real world, not from your source's internals, so that another source could produce the same key: `key=("ministerrat", url, file_number, office)`, `key=("side_job", member_id, employer, function)`. A key always overrides the facet-based identity. ### Summaries `summary_de` is required and shown to users; `summary_en` is optional but appreciated. Write neutral, factual one-liners ("X erhielt 2024 eine Spende von Y über 5.000 €"). Use `politrack.text.format_eur_de` for German amounts and `excerpt()` for quotes. ### Sensitive claims Anything about criminal or legal matters: `claim_type="legal_proceeding"`, `sensitive=True`, `review=True` unless the source is an official court/parliament record. The UI then shows the presumption-of-innocence notice and the claim waits for human review. ## 4. Incremental adapters and cursors For large or append-only sources, keep a cursor: ```python def start(self, ctx): since = ctx.params.get("since") or ctx.state.get("last_date") or "2024-01-01" yield Request(url=..., params={"from": since}, kind="page") def parse(self, ctx, response): ... ctx.state.set("last_date", newest_date_seen) # the only "write" parse() may do ``` `ctx.state` is persisted per adapter in the `state` table (and kept in memory during tests). ## 5. Multi-publisher adapters Feeds of many outlets, or an aggregator that points at original publishers: define one `SourceSpec` per publisher (register extra specs with `register_spec`) and put `meta={"source": ""}` on each request – the Fetcher then applies *that* publisher's policy (store_fulltext, throttle) and evidence is attributed to it. See `sources/news.py`. ## 6. Tests (required) 1. Capture a fixture from the live source (one request, honouring robots.txt): ```bash politrack fixture musterland.members "https://www.landtag-musterland.at/api/members?period=current" list_current --kind list ``` 2. Generate and **review** the golden output: ```bash UPDATE_GOLDEN=1 pytest tests/test_golden.py -k musterland git diff tests/fixtures/musterland.members/ ``` 3. Unit-test tricky helpers (regexes, date formats) in `tests/test_parsers.py`. Keep fixtures small (cut lists to a few entries) and free of personal data beyond public roles. ## 7. Changing an existing parser 1. Change `parse()`, bump `parser_version`. 2. `UPDATE_GOLDEN=1 pytest` and review the diff. 3. After deployment: `politrack reparse ` replays the raw archive – no refetching. At the end of a complete, error-free reparse, the adapter's claims that the new parser no longer produces are deleted (unless another source backs them), so a fix that changes a claim's identity leaves no stale duplicate. Two adapters sharing one `SourceSpec` should not emit the same claims, or a reparse of one removes the other's until its next run. ## 8. Checklist for the pull request - [ ] `SourceSpec` with honest `kind`, `trust`, `license`, `tos_notes` (legal checklist answered) - [ ] `parse()` is pure; counts skipped rows in `ctx.stats` - [ ] strong identifiers wherever the source has them - [ ] golden fixture + expected output, helper unit tests - [ ] entry in [docs/sources.md](../sources.md) - [ ] new claim types / identifier schemes documented in [data-model.md](../data-model.md) and the frontend (`ClaimDetails.tsx`, messages)