A Python client with requests and httpx

Build a small Python client with autocomplete, lookup, datavask, reverse geocoding, timeouts, error codes and 429 handling.

7 min Level: Intermediate ✓ Code tested

This guide gives you a complete Python client of about 120 lines that you can drop into your project. It covers the four calls most backends need: autocomplete, lookup by id, datavask (address cleansing) and reverse geocoding. It also handles timeouts, both error formats and responses with status 429.

There is no official Python package on PyPI at the moment. The code here only uses requests (or httpx for async code), so there are no extra dependencies to keep up to date.

Configuration

The client reads the key from DANADRESSE_API_KEY and the base URL from DANADRESSE_BASE_URL, defaulting to https://api.danadresse.dk. That lets you use a test key locally and a live key in production without changing code.

Shell#env
pip install requests
export DANADRESSE_API_KEY="dawa_live_din_nøgle_her"

Use a dawa_test_… key during development. It does not count against your account. See Test keys and the sandbox.

The whole client with requests

Save the file as danadresse_client.py. When you run it directly it makes six calls and prints the results (identifiers and comments are in Danish, matching the API's field names):

Python#requests-client
"""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)

Output:

Text#requests-output
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

How the parts work

One Session per process. requests.Session reuses the connection, so the second call skips a new TLS handshake. Create the client once, for example when your Django or Flask app starts, and share it.

Timeouts. (3.05, 10) means 3 seconds to connect and 10 seconds to read the response. Without a timeout a hanging connection can block a worker for minutes.

Datavask with A, B and C. /datavask/adresser returns one kategori at the top level and a list of resultater, best candidate first. Each result has adresse (the matched address), aktueladresse (the current version, which differs if the matched one is retired) and vaskeresultat with afstand (0 = exact), forskelle per field and parsetadresse. There is no score field, and the category exists only at the top level:

CategoryMeaningWhat the client should do
AExact match (case and punctuation aside)Store adresse.id
BExactly one safe match with small spelling differencesStore the id, and store the official text rather than what the user typed
CUncertain: several candidates or larger differencesDo not store automatically. Show the candidates and let the user pick

sikker in the example is true for A and B. "Strandvejen 100, 2900 Hellerup" gives A with one candidate. "Rentemestervej 8, 2. tv, 2400 København NV" gives C with seven candidates, because that number has no 2. tv. Use /datavask/adgangsadresser if you only need the building. betegnelse also reads floor and door (2. tv, st. th), lower-case house letters (12a) and aa/å spellings. If your form already has separate fields, you can send vejnavn, husnr, etage, dør and postnr instead. The datavask address has no coordinates, so look up /adresser/{id} if you need them.

Reverse geocoding. /adgangsadresser/reverse?x=<longitude>&y=<latitude> finds the nearest access address. Mind the order: x is the longitude. See Coordinates and srid if your coordinates are in UTM32 (srid=25832).

Errors and 429

Errors on the DAWA paths (/adresser, /autocomplete …) look like the first line below, while /api/v1/* puts the same fields under error. That holds for every error, 500 included. The text in title and message follows Accept-Language (Danish without the header):

JSON#error-shapes
{"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. …"}}

_fejl() reads code and details from both shapes, so your code can branch on e.code and never on the text. The most important codes:

StatuscodeAction
401missing_api_key, invalid_api_keyConfiguration error. Do not retry
403key_blocked, sandbox_scope, ip_not_allowedFix the key or plan. Do not retry
404not_foundThe id does not exist. Mark the address invalid
429rate_limit_per_minuteWait Retry-After seconds (1 to 60) and retry
429quota_exceededFree plan: the monthly quota and the grace buffer are used up. Paid plan: your budget cap is reached. No Retry-After. Never retry: stop, alert and show e.details["resets_at"] and e.details["upgrade_url"]
500internal_errorAn error on our side. Log it and try again later
503upstream_unavailableA public register is down. Wait Retry-After

e.details["reason"] on quota_exceeded says why: free_quota, budget_stop_at_quota or budget_cap_dkk. A free account that passes its monthly quota is still served up to 2,400 calls (20 % extra). Those responses carry X-Quota-Grace: true and X-Quota-Remaining, so log that header if you want an early warning. Calls refused with 429 are not counted or billed.

The client waits at most 30 seconds at a time and gives up after two retries. That keeps a web request within a sensible response time. In a batch job you can raise max_retries. The limits are described in Limits, quotas and 429.

Async variant with httpx

If you use FastAPI, aiohttp or another asyncio framework, use httpx.AsyncClient (pip install httpx). The structure is the same. Here three lookups run in parallel over one connection pool:

Python#httpx-client
"""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())

max_connections=10 caps how many calls run at once. That matters in batch jobs: send a thousand calls at once and you hit the per-minute limit and get 429 back. In FastAPI, create AsyncDanadresse in lifespan and close it on shutdown.

Store the id, not the text

When the user has picked a suggestion from autocomplete, store data.id with the order or customer. With the id you can always fetch the current address with api.adresse(id), even if the street is renamed. See Store the address id.

Next steps

Related guides

Markdown source Spotted an error? Tell us API-reference (Swagger)

Try it with a free key

2,000 calls/month on the free plan. No credit card.

Create account →