from __future__ import annotations import logging from dataclasses import dataclass import requests from app.ha.exceptions import ( HaAuthError, HaHttpError, HaTimeoutError, HaUnexpectedPayloadError, ) logger = logging.getLogger(__name__) @dataclass(frozen=True) class HaClientSettings: url: str token: str timeout_seconds: int = 10 class HaClient: def __init__(self, settings: HaClientSettings) -> None: self._settings = settings self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {settings.token}", "Content-Type": "application/json", }) def list_entities(self) -> list[dict[str, object]]: try: response = self._session.get( f"{self._settings.url}/api/states", timeout=self._settings.timeout_seconds, ) except requests.Timeout as exc: raise HaTimeoutError("Zeitüberschreitung beim Zugriff auf Home Assistant.") from exc except requests.RequestException as exc: raise HaHttpError( getattr(getattr(exc, "response", None), "status_code", 502), "Netzwerkfehler beim Zugriff auf Home Assistant.", ) from exc if response.status_code in (401, 403): raise HaAuthError( response.status_code, "Authentifizierung bei Home Assistant fehlgeschlagen.", ) try: response.raise_for_status() except requests.HTTPError as exc: raise HaHttpError( response.status_code, "Home Assistant meldet einen Fehler.", ) from exc try: payload = response.json() except ValueError as exc: raise HaUnexpectedPayloadError( "Antwort von Home Assistant ist kein gültiges JSON." ) from exc if not isinstance(payload, list): raise HaUnexpectedPayloadError( "Antwort von Home Assistant hat unerwartetes Format." ) return payload