Replikering, del 2: hold kopien opdateret med hændelser

GET /replikering/haendelser?entitet=adresse&sekvensnummerfra=207922000&sekvensnummertil=207940624&maxantal=1000
X-Api-Key: $DANADRESSE_API_KEY

import os

import requests

BASE = "https://api.danadresse.dk"
H = {"X-Api-Key": os.environ["DANADRESSE_API_KEY"]}

seneste = requests.get(f"{BASE}/replikering/senestesekvensnummer", headers=H, timeout=10).json()
print("seneste sekvensnummer:", seneste["sekvensnummer"])

r = requests.get(
    f"{BASE}/replikering/haendelser",
    params={"entitet": "adresse", "sekvensnummerfra": seneste["sekvensnummer"] - 20000, "maxantal": 3},
    headers=H,
    timeout=30,
)
r.raise_for_status()
for h in r.json():
    print(h["sekvensnummer"], h["operation"], h["tidspunkt"], h["data"]["id"])

{
  "txid": null,
  "sekvensnummer": 207922077,
  "tidspunkt": "2026-10-15T08:27:59.776Z",
  "operation": "delete",
  "data": {
    "id": "ff5063fe-432a-4b9a-ae18-7fcb0fbed0f4",
    "status": 2,
    "adgangsadresseid": "547c02a1-b1ce-4fc6-a40e-612c77d75490",
    "etage": null,
    "doer": null,
    "oprettet": "2026-10-15T10:27:59.776",
    "aendret": "2026-10-15T10:27:59.776"
  }
}

"""synk.py: hent hændelser siden sidste kørsel og opdater SQLite-kopien fra førsteudtrækket.

Kør fx hvert 15. minut fra cron. Sikker at køre igen efter et nedbrud: hændelser
anvendes som upserts, og cursoren gemmes i samme transaktion som ændringerne.
"""
import json
import os
import sqlite3
import time

import requests

BASE_URL = os.environ.get("DANADRESSE_BASE_URL", "https://api.danadresse.dk")
HEADERS = {"X-Api-Key": os.environ["DANADRESSE_API_KEY"]}
# Uden maxantal kommer hele intervallet fra..til i ét svar. Med maxantal deler du
# det op i sider, så én side aldrig bliver for stor (højst 50000).
MAXANTAL = 1000

# Samme rækkefølge og nøgler som i førsteudtrækket
ENTITETER = {
    "postnummer": ["nr"],
    "kommune": ["kode"],
    "navngivenvej": ["id"],
    "vejstykke": ["kommunekode", "vejkode"],
    "adgangsadresse": ["id"],
    "adresse": ["id"],
}

# Hændelser bruger enkelte andre feltnavne end udtrækket (DAWA-navne med æ/ø).
# Omdøb dem, så de lander i de samme nøgler i din tabel.
OMDOEB = {
    "dør": "doer",
    "ændret": "aendret",
    "ikrafttrædelsesdato": "ikrafttraedelsesdato",
    "nøjagtighed": "noejagtighed",
    "etrs89koordinat_øst": "etrs89_x",
    "etrs89koordinat_nord": "etrs89_y",
}


def get(path: str, **params) -> object:
    for forsoeg in range(3):
        r = requests.get(BASE_URL + path, params=params, headers=HEADERS, timeout=(5, 60))
        ra = r.headers.get("Retry-After", "")
        if r.status_code in (429, 503) and ra.isdigit() and forsoeg < 2:
            time.sleep(min(int(ra), 30))
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("unreachable")


def cursor(db: sqlite3.Connection, entitet: str) -> int:
    row = db.execute("SELECT sekvensnummer FROM replikering_cursor WHERE entitet = ?", (entitet,)).fetchone()
    if row is None:
        raise SystemExit(f"{entitet}: ingen cursor. Kør førsteudtrækket først.")
    return row[0]


def gem_cursor(db: sqlite3.Connection, entitet: str, sekvensnummer: int) -> None:
    db.execute("UPDATE replikering_cursor SET sekvensnummer = ? WHERE entitet = ?", (sekvensnummer, entitet))


def anvend(db: sqlite3.Connection, entitet: str, noegle: list[str], h: dict) -> None:
    data = {OMDOEB.get(k, k): v for k, v in h["data"].items()}
    if h["operation"] == "delete":
        # Behold rækken som nedlagt (2) eller henlagt (4), så gamle id'er kan slås op
        data["status"] = data.get("status") if data.get("status") in (2, 4) else 4
    pk = "-".join(str(data[k]) for k in noegle)
    # insert og update er begge en upsert. data kan være delvis, så flet den ind
    # i den eksisterende række (json_patch) i stedet for at overskrive den.
    db.execute(
        f"INSERT INTO {entitet} (pk, status, data) VALUES (?, ?, ?) "
        "ON CONFLICT(pk) DO UPDATE SET status = coalesce(excluded.status, status), "
        "data = json_patch(data, excluded.data)",
        (pk, data.get("status"), json.dumps(data, ensure_ascii=False)),
    )


def synk_entitet(db: sqlite3.Connection, entitet: str, noegle: list[str], til: int) -> int:
    fra = cursor(db, entitet) + 1
    if fra > til:
        return 0  # allerede hentet; flyt aldrig cursoren baglæns
    antal = 0
    while fra <= til:
        haendelser = get("/replikering/haendelser", entitet=entitet,
                         sekvensnummerfra=fra, sekvensnummertil=til, maxantal=MAXANTAL)
        for h in haendelser:
            anvend(db, entitet, noegle, h)
        if haendelser:
            gem_cursor(db, entitet, haendelser[-1]["sekvensnummer"])
        db.commit()  # ændringer og cursor i én transaktion
        antal += len(haendelser)
        if len(haendelser) < MAXANTAL:
            break  # færre end maxantal: intervallet er hentet
        fra = haendelser[-1]["sekvensnummer"] + 1
    gem_cursor(db, entitet, til)  # alt til og med `til` er hentet
    db.commit()
    return antal


def main() -> None:
    db = sqlite3.connect("adresser.db")
    til = get("/replikering/senestesekvensnummer")["sekvensnummer"]
    for entitet, noegle in ENTITETER.items():
        print(f"{entitet}: {synk_entitet(db, entitet, noegle, til)} hændelser (til {til})")


if __name__ == "__main__":
    main()

postnummer: 2 hændelser (til 207922555)
kommune: 0 hændelser (til 207922555)
navngivenvej: 1419 hændelser (til 207922555)
vejstykke: 1064 hændelser (til 207922555)
adgangsadresse: 2282 hændelser (til 207922555)
adresse: 428 hændelser (til 207922555)

# crontab: synkronisér hvert 15. minut, aldrig to kørsler på én gang
*/15 * * * * cd /srv/adresser && flock -n synk.lock python synk.py >> synk.log 2>&1