kundenplattform/anode_client.py

import os
from contextlib import contextmanager

import httpx


ANODE_URL = os.environ.get("TUXFLOTTE_ANODE_URL", "http://127.0.0.1:8080")
KUNDENPLATTFORM_TOKEN = os.environ.get("TUXFLOTTE_KUNDENPLATTFORM_TOKEN", "")


def anode_request(method: str, path: str, **kwargs) -> dict:
    response = httpx.request(
        method,
        ANODE_URL + path,
        headers={"Authorization": f"Bearer {KUNDENPLATTFORM_TOKEN}"},
        timeout=15,
        **kwargs,
    )
    response.raise_for_status()

    return response.json()


@contextmanager
def anode_stream(method: str, path: str):
    """
    Fuer den ISO-Download (Phase 5): der Browser hat kein
    KUNDENPLATTFORM_TOKEN, daher muss Kundenplattform als Proxy streamen
    statt einen direkten Link auf anode auszugeben. Kein Read-Timeout - eine
    mehrere GB grosse Datei ueber die WireGuard-Verbindung zu anode kann
    laenger dauern als der sonst genutzte 15s-Timeout erlaubt.
    """

    with httpx.stream(
        method,
        ANODE_URL + path,
        headers={"Authorization": f"Bearer {KUNDENPLATTFORM_TOKEN}"},
        timeout=httpx.Timeout(15, read=None),
    ) as response:
        yield response