kuma-strapper/backend/services/kuma_client.py

from typing import Optional
from dataclasses import dataclass, asdict
import requests

from config import get_config


@dataclass
class Monitor:
    """Uptime Kuma monitor configuration."""

    type: str  # http, tcp, ping, docker, keyword
    name: str
    url: Optional[str] = None  # For HTTP monitors
    hostname: Optional[str] = None  # For TCP/Ping monitors
    port: Optional[int] = None  # For TCP monitors
    interval: int = 60
    keyword: Optional[str] = None  # For keyword monitors
    docker_container: Optional[str] = None  # For Docker monitors
    docker_host: Optional[str] = None  # For Docker monitors
    retries: int = 3
    retry_interval: int = 60
    max_redirects: int = 10
    accepted_statuscodes: list[str] = None
    notification_id_list: Optional[list[int]] = None

    def __post_init__(self):
        if self.accepted_statuscodes is None:
            self.accepted_statuscodes = ["200-299"]

    def to_api_format(self) -> dict:
        """Convert to Uptime Kuma API format."""
        # Map our types to Kuma's type values
        type_map = {
            "http": "http",
            "tcp": "port",
            "ping": "ping",
            "docker": "docker",
            "keyword": "keyword",
        }

        data = {
            "type": type_map.get(self.type, self.type),
            "name": self.name,
            "interval": self.interval,
            "retries": self.retries,
            "retryInterval": self.retry_interval,
            "maxredirects": self.max_redirects,
            "accepted_statuscodes": self.accepted_statuscodes,
        }

        if self.url:
            data["url"] = self.url
        if self.hostname:
            data["hostname"] = self.hostname
        if self.port:
            data["port"] = self.port
        if self.keyword:
            data["keyword"] = self.keyword
        if self.docker_container:
            data["docker_container"] = self.docker_container
        if self.docker_host:
            data["docker_host"] = self.docker_host
        if self.notification_id_list:
            data["notificationIDList"] = self.notification_id_list

        return data


class UptimeKumaClient:
    """Client for Uptime Kuma REST API."""

    def __init__(self):
        config = get_config()
        self.base_url = config.uptime_kuma_url.rstrip("/")
        self.api_key = config.uptime_kuma_api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        })

    def _request(self, method: str, endpoint: str, **kwargs) -> dict:
        """Make an API request."""
        url = f"{self.base_url}/api{endpoint}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json() if response.content else {}

    def get_monitors(self) -> list[dict]:
        """Get all monitors."""
        try:
            result = self._request("GET", "/monitors")
            return result.get("monitors", [])
        except Exception as e:
            raise Exception(f"Failed to get monitors: {str(e)}")

    def get_monitor(self, monitor_id: int) -> dict:
        """Get a specific monitor."""
        try:
            result = self._request("GET", f"/monitors/{monitor_id}")
            return result.get("monitor", {})
        except Exception as e:
            raise Exception(f"Failed to get monitor {monitor_id}: {str(e)}")

    def create_monitor(self, monitor: Monitor) -> dict:
        """Create a new monitor."""
        try:
            data = monitor.to_api_format()
            result = self._request("POST", "/monitors", json=data)
            return result
        except Exception as e:
            raise Exception(f"Failed to create monitor: {str(e)}")

    def update_monitor(self, monitor_id: int, monitor: Monitor) -> dict:
        """Update an existing monitor."""
        try:
            data = monitor.to_api_format()
            result = self._request("PUT", f"/monitors/{monitor_id}", json=data)
            return result
        except Exception as e:
            raise Exception(f"Failed to update monitor {monitor_id}: {str(e)}")

    def delete_monitor(self, monitor_id: int) -> dict:
        """Delete a monitor."""
        try:
            result = self._request("DELETE", f"/monitors/{monitor_id}")
            return result
        except Exception as e:
            raise Exception(f"Failed to delete monitor {monitor_id}: {str(e)}")

    def pause_monitor(self, monitor_id: int) -> dict:
        """Pause a monitor."""
        try:
            result = self._request("POST", f"/monitors/{monitor_id}/pause")
            return result
        except Exception as e:
            raise Exception(f"Failed to pause monitor {monitor_id}: {str(e)}")

    def resume_monitor(self, monitor_id: int) -> dict:
        """Resume a paused monitor."""
        try:
            result = self._request("POST", f"/monitors/{monitor_id}/resume")
            return result
        except Exception as e:
            raise Exception(f"Failed to resume monitor {monitor_id}: {str(e)}")

    def get_status(self) -> dict:
        """Get Uptime Kuma status/info."""
        try:
            result = self._request("GET", "/status-page")
            return result
        except Exception as e:
            raise Exception(f"Failed to get status: {str(e)}")

    def test_connection(self) -> bool:
        """Test connection to Uptime Kuma."""
        try:
            self._request("GET", "/monitors")
            return True
        except Exception:
            return False


# Global client instance
_kuma_client: Optional[UptimeKumaClient] = None


def get_kuma_client() -> UptimeKumaClient:
    """Get the global Uptime Kuma client instance."""
    global _kuma_client
    if _kuma_client is None:
        _kuma_client = UptimeKumaClient()
    return _kuma_client