Refactor Uptime Kuma metrics retrieval

Author: tr4nt0r

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())

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",

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}
+    )

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()
+            }