File indexing completed on 2026-06-26 08:40:21
0001 """REST client for monitor /api/panda/* endpoints.
0002
0003 The alarm engine runs on the monitor host and uses the same public/local
0004 monitor base URL configured in config.toml.
0005 """
0006 from __future__ import annotations
0007
0008 import httpx
0009
0010
0011 class FetchError(RuntimeError):
0012 pass
0013
0014
0015 class Client:
0016 def __init__(self, base_url: str, timeout: int = 20):
0017 self.base_url = base_url.rstrip("/")
0018 self.timeout = timeout
0019
0020 def _get(self, path: str, params: dict | None = None) -> dict:
0021 url = f"{self.base_url}{path}"
0022 try:
0023 r = httpx.get(url, params=params, timeout=self.timeout, verify=True)
0024 except httpx.HTTPError as e:
0025 raise FetchError(f"request failed: {url}: {e}") from e
0026 if r.status_code >= 400:
0027 raise FetchError(f"{r.status_code} {r.reason_phrase} from {url}: {r.text[:200]}")
0028 try:
0029 return r.json()
0030 except ValueError as e:
0031 raise FetchError(f"non-json response from {url}: {r.text[:200]}") from e
0032
0033 def list_tasks(self, *, days: int = 1, status: str | None = None,
0034 username: str | None = None, taskname: str | None = None,
0035 processingtype: str | None = None,
0036 limit: int = 50, before_id: int | None = None) -> dict:
0037 params = {"days": days, "limit": limit}
0038 for k, v in (("status", status), ("username", username),
0039 ("taskname", taskname),
0040 ("processingtype", processingtype),
0041 ("before_id", before_id)):
0042 if v is not None:
0043 params[k] = v
0044 return self._get("/api/panda/tasks/", params)
0045
0046 def iter_all_tasks(self, **filters):
0047 """Paginate through all matching tasks. Yields task dicts."""
0048 before_id = None
0049 while True:
0050 batch = self.list_tasks(before_id=before_id, **filters)
0051 for item in batch.get("items", []):
0052 yield item
0053 if not batch.get("has_more"):
0054 return
0055 before_id = batch.get("next_before_id")
0056 if before_id is None:
0057 return
0058
0059 def get_task(self, jeditaskid: int) -> dict:
0060 return self._get(f"/api/panda/tasks/{jeditaskid}/")
0061
0062 def activity(self, *, days: int = 1) -> dict:
0063 return self._get("/api/panda/activity/", {"days": days})
