Python httpx async client — HTTP/2, retry og timeout til IoT-platforme

httpx er den moderne asynkrone HTTP-klient til Python. Her er de vigtigste mønstre til tredjepartsintegrationer i en IoT SaaS-platform.

---

AsyncClient vs. requests

# requests — synkron, blokerer event loop i async kontekst:
import requests
response = requests.get("https://api.dmi.dk/opendata/v1/metObs")  # BLOKERER

# httpx AsyncClient — korrekt async:
import httpx

async def fetch_dmi_data(api_key: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://dmigw.govcloud.dk/v2/climateData/collections/",
            headers={"X-Gravitee-Api-Key": api_key},
            timeout=10.0,
        )
        resp.raise_for_status()
        return resp.json()

# httpx features over aiohttp:
# - requests-kompatibel API (nem migration)
# - HTTP/2 support (httpx[http2])
# - Built-in retry via transport layers
# - HTTPX-specifik testclient kompatibel med ASGI

---

Persistent klient med connection pool

# server/src/integrations/http_client.py

import httpx
from contextlib import asynccontextmanager

_client: httpx.AsyncClient | None = None


def get_http_client() -> httpx.AsyncClient:
    """Returnér delt klient (singleton pr. process)."""
    if _client is None or _client.is_closed:
        raise RuntimeError("HTTP client ikke initialiseret")
    return _client


@asynccontextmanager
async def lifespan_http_client():
    """Bruges i FastAPI lifespan context manager."""
    global _client
    _client = httpx.AsyncClient(
        limits=httpx.Limits(
            max_connections=20,
            max_keepalive_connections=10,
            keepalive_expiry=30,
        ),
        timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0),
        http2=True,             # Kræver pip install httpx[http2]
        follow_redirects=True,
    )
    try:
        yield _client
    finally:
        await _client.aclose()

# FastAPI main.py:
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with lifespan_http_client():
        yield

---

Retry med tenacity

# server/src/integrations/brevo.py

import httpx
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log,
)
import structlog

log = structlog.get_logger()


@retry(
    retry=retry_if_exception_type((httpx.TransportError, httpx.TimeoutException)),
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=16),
    before_sleep=before_sleep_log(log, "warning"),
    reraise=True,
)
async def send_brevo_email(payload: dict) -> dict:
    """Send transaktionsemail via Brevo API med automatisk retry."""
    client = get_http_client()
    resp = await client.post(
        "https://api.brevo.com/v3/smtp/email",
        headers={
            "api-key": settings.brevo_api_key.get_secret_value(),
            "Content-Type": "application/json",
        },
        json=payload,
    )

    if resp.status_code == 429:
        # Rate limited — raise for tenacity retry:
        retry_after = int(resp.headers.get("Retry-After", "5"))
        raise httpx.HTTPStatusError(
            f"Rate limited, retry after {retry_after}s",
            request=resp.request, response=resp,
        )

    resp.raise_for_status()
    return resp.json()

---

DMI Klimagrid API integration

# server/src/integrations/dmi.py

from datetime import date, datetime

DMI_BASE = "https://dmigw.govcloud.dk/v2/climateData"


async def fetch_degree_days(
    zip_code: str,
    start: date,
    end: date,
) -> list[dict]:
    """Hent graddage-data fra DMI Klimagrid pr. postnummer."""
    client = get_http_client()

    # DMI bruger 10×10 km grid-celler — zip_code → celleid mapping:
    cell_id = ZIP_TO_DMI_CELL.get(zip_code)
    if not cell_id:
        raise ValueError(f"Ukendt postnummer: {zip_code}")

    resp = await client.get(
        f"{DMI_BASE}/collections/10kmGridValue/items",
        params={
            "cellId": cell_id,
            "parameter": "mean_temp",
            "datetime": f"{start}T00:00:00Z/{end}T23:59:59Z",
            "limit": 400,
        },
        headers={"X-Gravitee-Api-Key": settings.dmi_api_key.get_secret_value()},
        timeout=15.0,
    )
    resp.raise_for_status()

    features = resp.json()["features"]
    return [
        {
            "date": f["properties"]["from"][:10],
            "mean_temp_c": f["properties"]["value"],
            "heating_degree_days": max(0, 17.0 - f["properties"]["value"]),
        }
        for f in features
    ]

---

Mock i tests med respx

# server/tests/test_dmi_integration.py

import pytest
import respx
import httpx

@pytest.mark.asyncio
@respx.mock
async def test_fetch_degree_days_success():
    # Mock DMI API:
    respx.get(
        "https://dmigw.govcloud.dk/v2/climateData/collections/10kmGridValue/items"
    ).mock(return_value=httpx.Response(200, json={
        "features": [
            {
                "properties": {
                    "from": "2025-01-15T00:00:00Z",
                    "value": -2.5,   # -2.5°C → 19.5 HDD
                }
            }
        ]
    }))

    # Initialiser global client til test:
    async with lifespan_http_client():
        result = await fetch_degree_days("2100", date(2025, 1, 15), date(2025, 1, 15))

    assert len(result) == 1
    assert result[0]["heating_degree_days"] == pytest.approx(19.5)


@pytest.mark.asyncio
@respx.mock
async def test_brevo_rate_limit_retry():
    """Test at 429 rate limit trigger tenacity retry."""
    call_count = 0

    def side_effect(request):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            return httpx.Response(429, headers={"Retry-After": "1"})
        return httpx.Response(201, json={"messageId": "msg-123"})

    respx.post("https://api.brevo.com/v3/smtp/email").mock(side_effect=side_effect)

    async with lifespan_http_client():
        result = await send_brevo_email({"to": [{"email": "test@example.com"}]})

    assert result["messageId"] == "msg-123"
    assert call_count == 3   # 2 retries + 1 success

---

Rate limiting håndtering

# Respekter rate limits automatisk:

class RateLimitedClient:
    """Wrapper der automatisk venter ved 429-svar."""

    def __init__(self, client: httpx.AsyncClient):
        self._client = client

    async def request(self, method: str, url: str, **kwargs) -> httpx.Response:
        for attempt in range(5):
            resp = await self._client.request(method, url, **kwargs)
            if resp.status_code != 429:
                return resp

            retry_after = float(resp.headers.get("Retry-After", "2"))
            wait = min(retry_after * (2 ** attempt), 60.0)
            log.warning("Rate limited", url=url, wait_secs=wait)
            await asyncio.sleep(wait)

        resp.raise_for_status()
        return resp

---

Konklusion

httpx.AsyncClient med delt connection pool (singleton via lifespan) og tenacity-retry er standarden til tredjepartsintegrationer i FastAPI. HTTP/2 giver multiplexing på API'er som DMI og Brevo der understøtter det. respx mocker httpx-kald i tests uden at ramme rigtige API'er. Rate limit handling via Retry-After-header med exponential backoff forhindrer API-ban ved burst-adfærd.

Se Brevo transaktionsmails guide eller DMI graddage API guide.