Python httpx — async API-klient til IoT og SaaS

httpx er den moderne erstatning for requests til Python-applikationer der kræver async HTTP — og det er standard i FastAPI-projekter. Her er mønstrene der virker i produktion.

---

Basal AsyncClient

# server/src/integrations/http_client.py
import httpx
from contextlib import asynccontextmanager


@asynccontextmanager
async def get_client(base_url: str, timeout: float = 10.0):
    """
    Context manager der sikrer korrekt lukning af forbindelser.
    Brug som: async with get_client("https://api.example.com") as client:
    """
    async with httpx.AsyncClient(
        base_url=base_url,
        timeout=httpx.Timeout(
            connect=5.0,    # Forbindelsestimeout
            read=timeout,   # Læsetimeout
            write=5.0,      # Skrivetimeout
            pool=1.0,       # Ventetid på ledig forbindelse fra pool
        ),
        headers={"User-Agent": "mbus-gateway/1.0"},
        follow_redirects=True,
    ) as client:
        yield client


# Brug i endpoint:
async def fetch_dmi_data(zip_code: str) -> dict:
    async with get_client("https://dmigw.govcloud.dk") as client:
        response = await client.get(
            "/v2/climateData/collections/municipalityValue/items",
            params={"municipalityId": zip_code, "limit": 10},
            headers={"X-Gravitee-Api-Key": "..."},
        )
        response.raise_for_status()
        return response.json()

---

Persistent client (singleton pattern)

# server/src/integrations/persistent_client.py
# Genanvend én client på tværs af requests — connection pooling

import httpx
from functools import lru_cache


class BrevoEmailClient:
    """Persistent httpx-klient til Brevo email API."""

    def __init__(self, api_key: str):
        self._client = httpx.AsyncClient(
            base_url="https://api.brevo.com/v3",
            headers={
                "api-key": api_key,
                "Content-Type": "application/json",
            },
            timeout=httpx.Timeout(10.0),
            limits=httpx.Limits(
                max_connections=10,
                max_keepalive_connections=5,
                keepalive_expiry=30.0,
            ),
        )

    async def send_transactional(self, payload: dict) -> dict:
        response = await self._client.post("/smtp/email", json=payload)
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self._client.aclose()


# Lifespan-registrering i FastAPI:
# app.state.brevo = BrevoEmailClient(settings.brevo_api_key)
# ved shutdown: await app.state.brevo.close()

---

Tenacity retry — robust mod netværksfejl

# server/src/integrations/retry.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()

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}


def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in RETRYABLE_STATUS_CODES
    return isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.NetworkError))


@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=16),
    retry=retry_if_exception_type((httpx.ConnectError, httpx.ReadTimeout)),
    before_sleep=before_sleep_log(log, "warning"),
    reraise=True,
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str, **kwargs) -> httpx.Response:
    """
    GET med automatisk retry ved netværksfejl.
    Eksponentiel backoff: 2s → 4s → 8s → 16s
    """
    response = await client.get(url, **kwargs)
    if response.status_code in RETRYABLE_STATUS_CODES:
        response.raise_for_status()    # Trigger retry via HTTPStatusError
    return response

---

Stripe integration med httpx

# server/src/integrations/stripe_client.py
# Stripe Python SDK bruger requests internt — her er ren httpx alternativ

import httpx
import base64
from decimal import Decimal


class StripeHttpClient:
    BASE_URL = "https://api.stripe.com/v1"

    def __init__(self, secret_key: str):
        # Stripe bruger HTTP Basic Auth: nøgle som username, tomt password
        credentials = base64.b64encode(f"{secret_key}:".encode()).decode()
        self._client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Basic {credentials}",
                "Stripe-Version": "2024-04-10",
            },
            timeout=httpx.Timeout(30.0),
        )

    async def create_usage_record(
        self,
        subscription_item_id: str,
        quantity: int,
    ) -> dict:
        """Rapportér antal aktive enheder til Stripe metered billing."""
        response = await self._client.post(
            f"/subscription_items/{subscription_item_id}/usage_records",
            data={
                "quantity": str(quantity),
                "action": "set",
            },
        )
        response.raise_for_status()
        return response.json()

    async def get_upcoming_invoice(self, customer_id: str) -> Decimal:
        """Hent beløb for næste faktura."""
        response = await self._client.get(
            "/invoices/upcoming",
            params={"customer": customer_id},
        )
        response.raise_for_status()
        data = response.json()
        # Stripe returnerer øre — konvert til DKK
        return Decimal(data["amount_due"]) / Decimal("100")

---

Streaming response til store filer

# server/src/integrations/storage_client.py
# Download store filer fra Hetzner Object Storage uden at loade alt i RAM

import httpx
from pathlib import Path


async def download_firmware(url: str, destination: Path) -> None:
    """
    Stream download til fil — bruges ved OTA firmware-download til Pi.
    Loader aldrig hele filen i RAM.
    """
    async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            with destination.open("wb") as f:
                async for chunk in response.aiter_bytes(chunk_size=65536):
                    f.write(chunk)


async def upload_pdf(url: str, pdf_bytes: bytes, content_type: str = "application/pdf") -> None:
    """Upload PDF til S3-kompatibel storage."""
    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
        response = await client.put(
            url,
            content=pdf_bytes,
            headers={"Content-Type": content_type},
        )
        response.raise_for_status()

---

respx mocking i tests

# server/tests/test_dmi_client.py
import pytest
import respx
import httpx
from server.src.integrations.dmi import fetch_dmi_degree_days


@pytest.mark.asyncio
@respx.mock
async def test_fetch_dmi_degree_days_success():
    """Mock DMI API-kald uden netværk."""
    respx.get(
        "https://dmigw.govcloud.dk/v2/climateData/collections/municipalityValue/items"
    ).mock(
        return_value=httpx.Response(
            200,
            json={
                "features": [
                    {"properties": {"value": 12.5, "from": "2026-01-01T00:00:00Z"}}
                ]
            },
        )
    )

    result = await fetch_dmi_degree_days(zip_code="2800")

    assert result == 12.5


@pytest.mark.asyncio
@respx.mock
async def test_fetch_dmi_degree_days_404():
    """Håndter 404 fra DMI gracefully."""
    respx.get(respx.pattern.M("https://dmigw.govcloud.dk/**")).mock(
        return_value=httpx.Response(404)
    )

    result = await fetch_dmi_degree_days(zip_code="9999")

    assert result is None    # Ingen data → None, ikke exception


@pytest.mark.asyncio
async def test_fetch_dmi_retry(respx_mock):
    """pytest-httpx alternativ: Verificer at retry aktiveres ved 503."""
    import pytest_httpx

    # Første kald returnerer 503, andet kald returnerer 200
    # (bruges med pytest-httpx fixture i stedet for respx)
    pass    # Implementation afhænger af valgt mock-bibliotek

---

pytest-httpx alternativ

# server/tests/test_brevo_client.py
# pytest-httpx integrerer med httpx på transport-niveau

import pytest
from pytest_httpx import HTTPXMock
from server.src.integrations.brevo import BrevoEmailClient


@pytest.mark.asyncio
async def test_send_email_success(httpx_mock: HTTPXMock):
    httpx_mock.add_response(
        method="POST",
        url="https://api.brevo.com/v3/smtp/email",
        json={"messageId": "<test@brevo.com>"},
        status_code=201,
    )

    client = BrevoEmailClient(api_key="test-key")
    result = await client.send_transactional({
        "to": [{"email": "lejer@example.com"}],
        "subject": "Din årsafregning er klar",
        "htmlContent": "<p>Se din afregning her.</p>",
    })
    await client.close()

    assert result["messageId"] == "<test@brevo.com>"


@pytest.mark.asyncio
async def test_send_email_rate_limit(httpx_mock: HTTPXMock):
    httpx_mock.add_response(status_code=429)

    client = BrevoEmailClient(api_key="test-key")
    with pytest.raises(httpx.HTTPStatusError) as exc_info:
        await client.send_transactional({})
    await client.close()

    assert exc_info.value.response.status_code == 429

---

Timeout-strategi i produktion

# Timeout-konfiguration pr. use case:

TIMEOUTS = {
    # Hurtige interne API-kald (same datacenter)
    "internal": httpx.Timeout(connect=1.0, read=5.0, write=2.0, pool=0.5),

    # DMI Klimagrid (eksternt, men stabilt)
    "dmi": httpx.Timeout(connect=5.0, read=15.0, write=5.0, pool=1.0),

    # Brevo email (accepterer forsinkelse)
    "email": httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=1.0),

    # Stripe (kritisk — kortere timeout, tenacity retry)
    "stripe": httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=2.0),

    # OTA firmware download (stor fil, lang timeout)
    "ota_download": httpx.Timeout(connect=10.0, read=300.0, write=10.0, pool=2.0),
}

# Regel: Sæt ALTID explicit timeout — aldrig httpx.AsyncClient() uden timeout.
# Default er ingen timeout → produktionshang ved netværksfejl.

---

Konklusion

httpx med AsyncClient, persistent connection pooling og tenacity retry er grundopskriften til robuste integrations-klienter. respx og pytest-httpx giver fuld mock-kontrol i tests uden netværksafhængighed. Sæt altid explicit Timeout — ingen timeout er den hyppigste årsag til produktionshang.

Se Python Pydantic Settings guide eller FastAPI pagination guide.