En Python-klient med requests og httpx

pip install requests
export DANADRESSE_API_KEY="dawa_live_din_nøgle_her"

"""danadresse_client.py: en lille, komplet klient til Danadresse API (requests)."""
from __future__ import annotations

import os
import time
from dataclasses import dataclass
from typing import Any

import requests

BASE_URL = os.environ.get("DANADRESSE_BASE_URL", "https://api.danadresse.dk")


class DanadresseError(Exception):
    """Fejl fra API'et. `code` er den stabile kode, du kan forgrene på."""

    def __init__(self, status: int, code: str | None, message: str,
                 details: Any = None, retry_after: float | None = None):
        super().__init__(f"HTTP {status} {code}: {message}")
        self.status = status
        self.code = code
        self.message = message
        self.details = details          # fx resets_at og upgrade_url ved quota_exceeded
        self.retry_after = retry_after


@dataclass(frozen=True)
class Adresse:
    id: str
    betegnelse: str
    postnr: str | None
    kommunekode: str | None
    lon: float | None
    lat: float | None


@dataclass(frozen=True)
class Vask:
    kategori: str            # "A", "B" eller "C" (topniveau i svaret)
    adresse: Adresse | None  # bedste kandidat i den officielle stavning
    afstand: int | None      # vaskeresultat.afstand: 0 = præcis match
    kandidater: int          # antal resultater

    @property
    def sikker(self) -> bool:
        """A = præcis match, B = én sikker match med små stavefejl. C: lad brugeren vælge."""
        return self.kategori in ("A", "B") and self.adresse is not None


def _fejl(r: requests.Response) -> DanadresseError:
    try:
        body = r.json()
    except ValueError:
        body = {}
    if not isinstance(body, dict):
        body = {}
    # DAWA-stier: {"type","title","details","code"}  /api/v1: {"error":{"code","message"}}
    err = body["error"] if isinstance(body.get("error"), dict) else body
    ra = r.headers.get("Retry-After")
    return DanadresseError(
        r.status_code,
        err.get("code"),
        err.get("title") or err.get("message") or r.text[:200],
        err.get("details"),
        float(ra) if ra and ra.isdigit() else None,
    )


def _fra_dawa(d: dict, ag: dict) -> Adresse:
    lon, lat = (ag.get("adgangspunkt") or {}).get("koordinater") or (None, None)
    return Adresse(
        id=d["id"],
        betegnelse=d.get("adressebetegnelse") or "",
        postnr=(ag.get("postnummer") or {}).get("nr"),
        kommunekode=(ag.get("kommune") or {}).get("kode"),
        lon=lon,
        lat=lat,
    )


def _fra_vask(a: dict) -> Adresse:
    # Datavask-adressen har ingen koordinater eller kommunekode: hent /adresser/{id} til dem
    linje = f"{a['vejnavn']} {a['husnr']}"
    enhed = " ".join(filter(None, [f"{a['etage']}." if a.get("etage") else None, a.get("dør")]))
    if enhed:
        linje += ", " + enhed
    return Adresse(
        id=a["id"],
        betegnelse=f"{linje}, {a['postnr']} {a['postnrnavn']}",
        postnr=a.get("postnr"),
        kommunekode=None,
        lon=None,
        lat=None,
    )


class DanadresseClient:
    def __init__(self, api_key: str | None = None, base_url: str = BASE_URL,
                 timeout: tuple[float, float] = (3.05, 10), max_retries: int = 2):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout          # (connect, read) i sekunder
        self.max_retries = max_retries
        self.session = requests.Session()  # genbruger TCP/TLS-forbindelser
        self.session.headers["X-Api-Key"] = api_key or os.environ["DANADRESSE_API_KEY"]

    def _get(self, path: str, **params: Any) -> Any:
        for forsoeg in range(self.max_retries + 1):
            try:
                r = self.session.get(self.base_url + path, params=params, timeout=self.timeout)
            except (requests.ConnectionError, requests.Timeout):
                if forsoeg == self.max_retries:
                    raise
                time.sleep(2 ** forsoeg)
                continue
            if r.ok:
                return r.json()
            err = _fejl(r)
            # 429 med Retry-After (minutgrænse) og 503 (register nede) kan prøves igen.
            # quota_exceeded har ingen Retry-After: kvoten eller budgetloftet er nået.
            if r.status_code in (429, 503) and err.retry_after and forsoeg < self.max_retries:
                time.sleep(min(err.retry_after, 30))
                continue
            raise err
        raise RuntimeError("unreachable")

    def autocomplete(self, q: str, per_side: int = 5) -> list[dict]:
        return self._get("/autocomplete", q=q, per_side=per_side)

    def adresse(self, adresse_id: str) -> Adresse:
        d = self._get(f"/adresser/{adresse_id}")
        return _fra_dawa(d, d.get("adgangsadresse") or {})

    def vask(self, betegnelse: str) -> Vask:
        svar = self._get("/datavask/adresser", betegnelse=betegnelse)
        res = svar.get("resultater") or []
        bedste = res[0] if res else {}
        # aktueladresse = den gældende version (afviger, hvis den fundne er nedlagt)
        a = bedste.get("aktueladresse") or bedste.get("adresse")
        return Vask(
            kategori=svar.get("kategori", "C"),
            adresse=_fra_vask(a) if a else None,
            afstand=(bedste.get("vaskeresultat") or {}).get("afstand"),
            kandidater=len(res),
        )

    def reverse(self, lon: float, lat: float) -> Adresse:
        d = self._get("/adgangsadresser/reverse", x=lon, y=lat)
        return _fra_dawa(d, d)


if __name__ == "__main__":
    api = DanadresseClient()
    for f in api.autocomplete("Rådhuspladsen 1, 4200", per_side=3):
        print("forslag:", f["forslagstekst"])
    a = api.adresse("0a3f50ae-b4ee-32b8-e044-0003ba298018")
    print("opslag:", a.betegnelse, a.lon, a.lat)
    for tekst in ("Strandvejen 100, 2900 Hellerup", "Rentemestervej 8, 2. tv, 2400 København NV"):
        v = api.vask(tekst)
        print("datavask:", v.kategori, v.kandidater,
              v.adresse.betegnelse if v.sikker else "lad brugeren vælge")
    r = api.reverse(12.5683, 55.6761)
    print("reverse:", r.betegnelse)
    try:
        api.adresse("00000000-0000-0000-0000-000000000000")
    except DanadresseError as e:
        print("fejl:", e.status, e.code)

forslag: Rådhuspladsen 1, 4200 Slagelse
forslag: Rådhuspladsen 1, st. tv, 4200 Slagelse
forslag: Rådhuspladsen 1, st. th, 4200 Slagelse
opslag: Rådhuspladsen 1, st. th, 4200 Slagelse 11.35623868 55.40464424
datavask: A 1 Strandvejen 100, 2900 Hellerup
datavask: C 7 lad brugeren vælge
reverse: Rådhuspladsen 3, 1550 København V
fejl: 404 not_found

{"type": "RateLimitExceededError", "title": "Too many requests — the limit is 360 per minute. Try again in 12 seconds.", "details": {"limit_per_minute": 360, "reset_at": 1793620860, "retry_after": 12}, "code": "rate_limit_per_minute"}
{"detail": {"error": {"code": "sandbox_scope", "message": "The sandbox/demo key is limited to basic address lookups. …"}}, "error": {"code": "sandbox_scope", "message": "The sandbox/demo key is limited to basic address lookups. …"}}

"""Asynkron variant med httpx: én delt AsyncClient, timeouts og Retry-After."""
from __future__ import annotations

import asyncio
import os
from dataclasses import dataclass
from typing import Any

import httpx

BASE_URL = os.environ.get("DANADRESSE_BASE_URL", "https://api.danadresse.dk")


@dataclass(frozen=True)
class Adresse:
    id: str
    betegnelse: str
    postnr: str | None
    lon: float | None
    lat: float | None


class DanadresseError(Exception):
    def __init__(self, status: int, code: str | None, message: str):
        super().__init__(f"HTTP {status} {code}: {message}")
        self.status, self.code, self.message = status, code, message


class AsyncDanadresse:
    def __init__(self, api_key: str | None = None, max_retries: int = 2):
        self.max_retries = max_retries
        self.client = httpx.AsyncClient(
            base_url=BASE_URL,
            headers={"X-Api-Key": api_key or os.environ["DANADRESSE_API_KEY"]},
            timeout=httpx.Timeout(10.0, connect=3.0),
            limits=httpx.Limits(max_connections=10),
        )

    async def __aenter__(self) -> "AsyncDanadresse":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.client.aclose()

    async def get(self, path: str, **params: Any) -> Any:
        for forsoeg in range(self.max_retries + 1):
            try:
                r = await self.client.get(path, params=params)
            except httpx.TransportError:  # timeout, DNS, afbrudt forbindelse
                if forsoeg == self.max_retries:
                    raise
                await asyncio.sleep(2 ** forsoeg)
                continue
            if r.is_success:
                return r.json()
            try:
                body = r.json()
            except ValueError:
                body = {}
            err = body.get("error") if isinstance(body.get("error"), dict) else body
            ra = r.headers.get("Retry-After", "")
            if r.status_code in (429, 503) and ra.isdigit() and forsoeg < self.max_retries:
                await asyncio.sleep(min(int(ra), 30))
                continue
            raise DanadresseError(r.status_code, err.get("code"),
                                  err.get("title") or err.get("message") or "")
        raise RuntimeError("unreachable")

    async def adresse(self, adresse_id: str) -> Adresse:
        d = await self.get(f"/adresser/{adresse_id}")
        ag = d["adgangsadresse"]
        lon, lat = ag["adgangspunkt"]["koordinater"]
        return Adresse(d["id"], d["adressebetegnelse"], ag["postnummer"]["nr"], lon, lat)


async def main() -> None:
    ids = [
        "0a3f50ae-b4ee-32b8-e044-0003ba298018",
        "762a9a1f-a5fb-4e47-b1b2-9fa94026bc80",
        "fec4202d-1015-425d-a3f5-5d21d03194b6",
    ]
    async with AsyncDanadresse() as api:
        # Tre opslag parallelt over den samme forbindelsespulje
        for a in await asyncio.gather(*(api.adresse(i) for i in ids)):
            print(a.postnr, a.betegnelse)


asyncio.run(main())