50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from . import __version__
|
|
|
|
USER_AGENT = f"fluxer-admin-alerts/{__version__}"
|
|
|
|
|
|
class FluxerAPIError(Exception):
|
|
pass
|
|
|
|
|
|
class FluxerAPI:
|
|
def __init__(self, base_url: str, api_key: str, timeout: float) -> None:
|
|
self._base_url = base_url
|
|
self._timeout = timeout
|
|
self._session = requests.Session()
|
|
self._session.headers.update(
|
|
{
|
|
"Authorization": f"Admin {api_key}",
|
|
"Accept": "application/json",
|
|
"User-Agent": USER_AGENT,
|
|
}
|
|
)
|
|
|
|
def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
|
|
try:
|
|
resp = self._session.get(f"{self._base_url}{path}", params=params, timeout=self._timeout)
|
|
except requests.RequestException as exc:
|
|
raise FluxerAPIError(f"GET {path} failed: {type(exc).__name__}") from exc
|
|
|
|
if not resp.ok:
|
|
raise FluxerAPIError(f"GET {path} returned {resp.status_code} ({_error_code(resp)})")
|
|
try:
|
|
return resp.json()
|
|
except ValueError as exc:
|
|
raise FluxerAPIError(f"GET {path} returned invalid JSON") from exc
|
|
|
|
|
|
def _error_code(resp: requests.Response) -> str:
|
|
try:
|
|
body = resp.json()
|
|
except ValueError:
|
|
return "no error body"
|
|
if isinstance(body, dict):
|
|
return str(body.get("code", "unknown error"))
|
|
return "unknown error"
|