diff --git a/README.md b/README.md index 66c1781..b122d81 100644 --- a/README.md +++ b/README.md @@ -16,17 +16,16 @@ import aiohttp from pythonkuma import UptimeKuma -URL = "" -USERNAME = "" -PASSWORD = "" +URL = "https://uptime.exampe.com" +API_KEY = "api_key" async def main(): async with aiohttp.ClientSession() as session: - uptime_kuma = UptimeKuma(session, URL, USERNAME, PASSWORD) - response = await uptime_kuma.async_get_monitors() - print(response.data) + uptime_kuma = UptimeKuma(session, URL, API_KEY) + response = await uptime_kuma.metrics() + print(response) asyncio.run(main()) diff --git a/pythonkuma/__init__.py b/pythonkuma/__init__.py index e6a4415..4823943 100644 --- a/pythonkuma/__init__.py +++ b/pythonkuma/__init__.py @@ -1,7 +1,11 @@ """Python API wrapper for Uptime Kuma.""" -from .exceptions import UptimeKumaAuthenticationException, UptimeKumaConnectionException, UptimeKumaException -from .models import MonitorStatus, MonitorType, UptimeKumaApiResponse, UptimeKumaMonitor +from .exceptions import ( + UptimeKumaAuthenticationException, + UptimeKumaConnectionException, + UptimeKumaException, +) +from .models import MonitorStatus, MonitorType, UptimeKumaMonitor from .uptimekuma import UptimeKuma __version__ = "0.0.0rc0" @@ -10,7 +14,6 @@ "MonitorStatus", "MonitorType", "UptimeKuma", - "UptimeKumaApiResponse", "UptimeKumaAuthenticationException", "UptimeKumaConnectionException", "UptimeKumaException", diff --git a/pythonkuma/models.py b/pythonkuma/models.py index 9dc2666..240c888 100644 --- a/pythonkuma/models.py +++ b/pythonkuma/models.py @@ -1,13 +1,12 @@ -"""Uptime Kuma models""" +"""Uptime Kuma models.""" from __future__ import annotations from dataclasses import dataclass, field from enum import IntEnum, StrEnum -from typing import Any +from typing import Self from mashumaro import DataClassDictMixin -from prometheus_client.parser import text_string_to_metric_families as parser class MonitorStatus(IntEnum): @@ -44,6 +43,12 @@ class MonitorType(StrEnum): RADIUS = "radius" REDIS = "redis" TAILSCALE_PING = "tailscale-ping" + UNKNOWN = "unknown" + + @classmethod + def _missing_(cls, _: object) -> Self: + """Handle new and unknown monitor types.""" + return cls.UNKNOWN @dataclass @@ -57,46 +62,16 @@ class UptimeKumaMonitor(UptimeKumaBaseModel): monitor_cert_days_remaining: int monitor_cert_is_valid: bool - monitor_hostname: str | None = field(metadata={"deserialize": lambda v: None if v == "null" else v}) + monitor_hostname: str | None = field( + metadata={"deserialize": lambda v: None if v == "null" else v} + ) monitor_name: str - monitor_port: str | None = field(metadata={"deserialize": lambda v: None if v == "null" else v}) + monitor_port: str | None = field( + metadata={"deserialize": lambda v: None if v == "null" else v} + ) monitor_response_time: int = 0 monitor_status: MonitorStatus monitor_type: MonitorType = MonitorType.HTTP - monitor_url: str | None = field(metadata={"deserialize": lambda v: None if v == "null" else v}) - - -@dataclass -class UptimeKumaApiResponse(UptimeKumaBaseModel): - """API response model for Uptime Kuma.""" - - _method: str | None = None - _api_path: str | None = None - data: list[UptimeKumaMonitor] | None = None - - @staticmethod - def from_prometheus(data: dict[str, Any]) -> UptimeKumaApiResponse: - """Generate object from json.""" - obj: dict[str, Any] = {} - monitors = [] - - for key, value in data.items(): - if hasattr(UptimeKumaApiResponse, key): - obj[key] = value - - parsed = parser(data["monitors"]) - for family in parsed: - for sample in family.samples: - if sample.name.startswith("monitor"): - existed = next( - (i for i, x in enumerate(monitors) if x["monitor_name"] == sample.labels["monitor_name"]), - None, - ) - if existed is None: - temp = {**sample.labels, sample.name: sample.value} - monitors.append(temp) - else: - monitors[existed][sample.name] = sample.value - obj["data"] = [UptimeKumaMonitor.from_dict(monitor) for monitor in monitors] - - return UptimeKumaApiResponse(**obj) + monitor_url: str | None = field( + metadata={"deserialize": lambda v: None if v == "null" else v} + ) diff --git a/pythonkuma/uptimekuma.py b/pythonkuma/uptimekuma.py index 4d0dd2f..959eeb5 100644 --- a/pythonkuma/uptimekuma.py +++ b/pythonkuma/uptimekuma.py @@ -1,23 +1,76 @@ """Uptime Kuma client.""" -from aiohttp import ClientSession +from http import HTTPStatus +from typing import Any + +from aiohttp import ( + BasicAuth, + ClientError, + ClientResponseError, + ClientSession, + ClientTimeout, +) +from prometheus_client.parser import text_string_to_metric_families from yarl import URL -from .decorator import api_request -from .models import UptimeKumaApiResponse +from .exceptions import UptimeKumaAuthenticationException, UptimeKumaConnectionException +from .models import UptimeKumaMonitor class UptimeKuma: - """This class is used to get information from Uptime Kuma.""" + """Uptime Kuma client.""" - def __init__(self, session: ClientSession, base_url: URL | str, username: str, password: str) -> None: - """Initialize""" - self.monitors = [] + def __init__( + self, + session: ClientSession, + base_url: URL | str, + api_key: str | None = None, + timeout: float | None = None, + ) -> None: + """Initialize the Uptime Kuma client.""" self._base_url = base_url if isinstance(base_url, URL) else URL(base_url) - self._username = username - self._password = password - self._session: ClientSession = session - @api_request("metrics") - async def async_get_monitors(self, **kwargs) -> UptimeKumaApiResponse: - """Get monitors from API.""" + self._auth = BasicAuth("", api_key) if api_key else None + + self._timeout = ClientTimeout(total=timeout or 10) + self._session = session + + async def metrics(self) -> list[UptimeKumaMonitor]: + """Retrieve metrics from Uptime Kuma.""" + url = self._base_url / "metrics" + + try: + request = await self._session.get( + url, auth=self._auth, timeout=self._timeout + ) + request.raise_for_status() + except ClientResponseError as e: + if e.status is HTTPStatus.UNAUTHORIZED: + msg = "Authentication failed for %s" + raise UptimeKumaAuthenticationException(msg, str(url)) from e + msg = "Request for %s failed with status code %s" + raise UptimeKumaConnectionException(msg, str(url), e.status) from e + except TimeoutError as e: + msg = "Request timeout for %s" + raise UptimeKumaConnectionException(msg, str(url)) from e + except ClientError as e: + raise UptimeKumaConnectionException from e + else: + parsed = text_string_to_metric_families(await request.text()) + + monitors: dict[str, dict[str, Any]] = {} + for metric in parsed: + if not metric.name.startswith("monitor"): + continue + for sample in metric.samples: + if not (monitor_name := sample.labels.get("monitor_name")): + continue + + monitors.setdefault(monitor_name, sample.labels).update( + {sample.name: sample.value} + ) + + return { + key: UptimeKumaMonitor.from_dict(value) + for key, value in monitors.items() + }