politics-tracking/docs/contributing/adapters.md

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 §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 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/<adapter name>/
  <case>.request.json       captured request (url, kind, meta)
  <case>.html|json|xml|bin  captured response body
  <case>.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:

"""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"             # <source>.<dataset>; 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:

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).

Referring to people and organisations

Never look up ids yourself. Describe what you know and let the Loader resolve it:

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:

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": "<slug>"} 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):

   politrack fixture musterland.members "https://www.landtag-musterland.at/api/members?period=current" list_current --kind list
   

2. Generate and review the golden output:

   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 <adapter> 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
• new claim types / identifier schemes documented in data-model.md and the frontend (ClaimDetails.tsx, messages)