from __future__ import annotations import logging from dataclasses import dataclass from typing import Any 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, Any]]: try: response = self._session.get( f"{self._settings.url}/api/states", timeout=self._settings.timeout_seconds, ) except requests.Timeout as exc: raise HaTimeoutError("Home Assistant request timed out.") from exc except requests.RequestException as exc: raise HaHttpError(status_code=502, message="Home Assistant request failed.") from exc if response.status_code in {401, 403}: raise HaAuthError( status_code=response.status_code, message="Home Assistant authentication failed.", ) try: response.raise_for_status() except requests.HTTPError as exc: raise HaHttpError( status_code=response.status_code, message="Home Assistant returned an HTTP error.", ) from exc try: payload = response.json() except ValueError as exc: raise HaUnexpectedPayloadError("Home Assistant returned invalid JSON.") from exc if not isinstance(payload, list): raise HaUnexpectedPayloadError("Home Assistant states response must be a list.") return payload