Replikering, del 1: hent et fuldt udtræk ind i din database

{"sekvensnummer": 207940624, "tidspunkt": "2026-11-02T12:23:10.000Z"}

import json
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"])

# ndjson: én række pr. linje. Uden den får du ét stort JSON-array.
with requests.get(f"{BASE}/replikering/udtraek", params={"entitet": "postnummer", "ndjson": ""},
                  headers=H, stream=True, timeout=(5, 30)) as r:
    r.raise_for_status()
    for i, linje in enumerate(r.iter_lines()):
        print(json.loads(linje))
        if i == 2:
            break

{"id": "000021c5-e9ee-411d-b2d8-ec9161780ccd", "status": 1, "adgangsadresseid": "0a3f5086-e9cd-32b8-e044-0003ba298018", "etage": "2", "doer": "tv", "oprettet": "2018-08-02T12:47:31", "aendret": "2018-08-02T12:47:31"}

"""foerste_udtraek.py: fuldt udtræk af adressedata ind i SQLite.

    python foerste_udtraek.py                      # alle entiteter i ENTITETER
    python foerste_udtraek.py postnummer kommune   # kun disse
"""
import json
import os
import sqlite3
import sys
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"]}

# Rækkefølge: det, der bliver refereret til, før det, der refererer
ENTITETER = ["postnummer", "kommune", "navngivenvej", "vejstykke", "adgangsadresse", "adresse"]
BATCH = 5000


def opret_tabeller(db: sqlite3.Connection, entiteter: list[str]) -> None:
    for e in entiteter:
        # pk = primærnøglen som tekst; data = hele rækken som JSON
        db.execute(f"CREATE TABLE IF NOT EXISTS {e} (pk TEXT PRIMARY KEY, status INTEGER, data TEXT NOT NULL)")
    db.execute("CREATE TABLE IF NOT EXISTS replikering_cursor "
               "(entitet TEXT PRIMARY KEY, sekvensnummer INTEGER NOT NULL)")
    db.commit()


def hent_json(path: str) -> dict:
    r = requests.get(BASE_URL + path, headers=HEADERS, timeout=(5, 30))
    r.raise_for_status()
    return r.json()


def hent_udtraek(db: sqlite3.Connection, entitet: str, noegle: list[str]) -> int:
    sql = (f"INSERT INTO {entitet} (pk, status, data) VALUES (?, ?, ?) "
           "ON CONFLICT(pk) DO UPDATE SET status = excluded.status, data = excluded.data")
    antal = 0
    rows: list[tuple] = []
    # stream=True: rækkerne skrives, mens de kommer, i stedet for at ligge i hukommelsen.
    # Read-timeout gælder mellem to bidder af svaret, ikke hele overførslen.
    # ndjson: én række pr. linje, så iter_lines() kan læse rækkerne én ad gangen.
    with requests.get(f"{BASE_URL}/replikering/udtraek", params={"entitet": entitet, "ndjson": ""},
                      headers=HEADERS, stream=True, timeout=(5, 120)) as r:
        r.raise_for_status()
        for linje in r.iter_lines():
            if not linje:
                continue
            raekke = json.loads(linje)
            pk = "-".join(str(raekke[k]) for k in noegle)
            rows.append((pk, raekke.get("status"), json.dumps(raekke, ensure_ascii=False)))
            if len(rows) >= BATCH:
                db.executemany(sql, rows)
                db.commit()
                antal += len(rows)
                rows.clear()
    if rows:
        db.executemany(sql, rows)
        db.commit()
        antal += len(rows)
    return antal


def main(entiteter: list[str]) -> None:
    db = sqlite3.connect("adresser.db")
    opret_tabeller(db, entiteter)
    model = hent_json("/replikering/datamodel")  # primærnøgle pr. entitet

    # 1. Gem sekvensnummeret FØR udtrækket. Ændringer under udtrækket hentes
    #    bagefter som hændelser og overskriver idempotent.
    start = hent_json("/replikering/senestesekvensnummer")["sekvensnummer"]
    print("start-sekvensnummer:", start)

    for entitet in entiteter:
        t0 = time.monotonic()
        antal = hent_udtraek(db, entitet, model[entitet]["key"])
        # 2. Cursor pr. entitet: herfra skal hændelser hentes næste gang
        db.execute("INSERT INTO replikering_cursor (entitet, sekvensnummer) VALUES (?, ?) "
                   "ON CONFLICT(entitet) DO UPDATE SET sekvensnummer = excluded.sekvensnummer",
                   (entitet, start))
        db.commit()
        print(f"{entitet}: {antal} rækker på {time.monotonic() - t0:.0f} s")


if __name__ == "__main__":
    main(sys.argv[1:] or ENTITETER)

start-sekvensnummer: 207940624
postnummer: 1108 rækker på 0 s
kommune: 99 rækker på 1 s
navngivenvej: 112484 rækker på 31 s
vejstykke: 113877 rækker på 29 s

CREATE TABLE adresse (
  pk     text PRIMARY KEY,
  status smallint,
  data   jsonb NOT NULL
);
CREATE INDEX adresse_adgangsadresseid ON adresse ((data->>'adgangsadresseid'));

-- Én række pr. linje i udtrækket (psycopg: cur.executemany(sql, rows))
INSERT INTO adresse (pk, status, data) VALUES (%s, %s, %s::jsonb)
ON CONFLICT (pk) DO UPDATE
  SET status = EXCLUDED.status,
      data   = adresse.data || EXCLUDED.data;

CREATE TABLE replikering_cursor (entitet text PRIMARY KEY, sekvensnummer bigint NOT NULL);