Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 35 additions & 6 deletions pythonkuma/uptimekuma.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,19 @@ def __init__(
api_key: str | None = None,
timeout: float | None = None,
) -> None:
"""Initialize the Uptime Kuma client."""
"""Initialize the Uptime Kuma client.

Parameters
----------
session : ClientSession
An aiohttp ClientSession instance
base_url : URL or str
The base URL of the Uptime Kuma server
api_key : str or None, optional
API key for authentication (default is None).
timeout : float or None, optional
Request timeout in seconds (default is 10 seconds if not specified).
"""
self._base_url = base_url if isinstance(base_url, URL) else URL(base_url)

self._auth = BasicAuth("", api_key) if api_key else None
Expand All @@ -36,7 +48,27 @@ def __init__(
self._session = session

async def metrics(self) -> dict[str, UptimeKumaMonitor]:
"""Retrieve metrics from Uptime Kuma."""
"""Retrieve metrics from Uptime Kuma.

Fetches and parses Prometheus-style metrics from the Uptime Kuma API endpoint,
extracting monitor-related metrics and returning them as a dictionary of
UptimeKumaMonitor objects keyed by monitor name.

Returns
-------
dict[str, UptimeKumaMonitor]
A dictionary mapping monitor names to their corresponding UptimeKumaMonitor
objects.

Raises
------
UptimeKumaAuthenticationException
If authentication with the Uptime Kuma API fails.
UptimeKumaConnectionException
If there is a connection error, timeout, or other client error during the
request.
"""
monitors: dict[str, dict[str, Any]] = {}
url = self._base_url / "metrics"

try:
Expand All @@ -56,10 +88,7 @@ async def metrics(self) -> dict[str, UptimeKumaMonitor]:
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:
for metric in text_string_to_metric_families(await request.text()):
if not metric.name.startswith("monitor"):
continue
for sample in metric.samples:
Expand Down
Loading