53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
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__)
|
|
|
|
|
|
class HaClient:
|
|
def __init__(self, settings) -> 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):
|
|
url = f"{self._settings.url}/api/states"
|
|
try:
|
|
response = self._session.get(
|
|
url,
|
|
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(exc.response, "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 |