kadence/desktop/kadence_desktop/api.py

"""HTTP client for the backend.

Deliberately stdlib-only (`urllib`): the desktop app should start with the
system Python on a fresh machine, and a pomodoro client does not need an async
HTTP stack to call eight endpoints.
"""

from __future__ import annotations

import json
import urllib.error
import urllib.parse
import urllib.request
from typing import Any


class ApiError(RuntimeError):
    """A call failed. The message is meant to be shown in the UI."""

    def __init__(self, message: str, status: int = 0) -> None:
        super().__init__(message)
        self.status = status


class Api:
    def __init__(self, base_url: str, token: str = "", timeout: float = 10.0) -> None:
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.timeout = timeout

    def configure(self, base_url: str, token: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.token = token

    # ---- plumbing ----------------------------------------------------------

    def _request(self, method: str, path: str, payload: Any | None = None) -> Any:
        url = f"{self.base_url}{path}"
        data = None
        headers = {"Accept": "application/json"}
        if payload is not None:
            data = json.dumps(payload).encode("utf-8")
            headers["Content-Type"] = "application/json"
        if self.token:
            headers["X-API-Token"] = self.token

        request = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                body = response.read()
                return json.loads(body) if body else None
        except urllib.error.HTTPError as exc:
            detail = _detail(exc)
            if exc.code == 401:
                hint = (
                    "Set the API token in Settings → Backend."
                    if self.token
                    else "No API token configured — add it in Settings → Backend."
                )
                raise ApiError(f"{detail} {hint}", exc.code) from exc
            raise ApiError(detail, exc.code) from exc
        except (urllib.error.URLError, TimeoutError) as exc:
            raise ApiError(
                f"Cannot reach the backend at {self.base_url} ({exc}). "
                "Check the server URL in Settings → Backend, or is "
                "`docker compose up` running locally?"
            ) from exc
        except json.JSONDecodeError as exc:
            raise ApiError(f"The backend sent something that is not JSON: {exc}") from exc

    # ---- endpoints ---------------------------------------------------------

    def health(self) -> dict:
        return self._request("GET", "/api/health")

    def timer_state(self) -> dict:
        return self._request("GET", "/api/timer/state")

    def start(self, kind: str | None = None, task_id: int | None = None, minutes: int | None = None) -> dict:
        return self._request(
            "POST", "/api/timer/start", {"kind": kind, "task_id": task_id, "minutes": minutes}
        )

    def pause(self) -> dict:
        return self._request("POST", "/api/timer/pause", {})

    def resume(self) -> dict:
        return self._request("POST", "/api/timer/resume", {})

    def complete(self, auto_next: bool = True, session_id: int | None = None) -> dict:
        """`session_id` names the interval we think we are finishing.

        The server ignores the call if something else is running by then, which
        is what keeps two clients hitting zero at the same moment from eating
        the break the first one's `auto_next` just started.
        """
        return self._request(
            "POST",
            "/api/timer/complete",
            {"auto_next": auto_next, "session_id": session_id},
        )

    def abandon(self, reason: str | None = None) -> dict:
        return self._request("POST", "/api/timer/abandon", {"reason": reason})

    def extend(self, minutes: int = 5) -> dict:
        return self._request("POST", "/api/timer/extend", {"minutes": minutes})

    def interrupt(self) -> dict:
        return self._request("POST", "/api/timer/interrupt", {})

    def tasks(self) -> dict:
        """The list, read for one fact: which task is pinned.

        There is no Tasks page here any more — the list is the web app's job.
        `Controller._on_tasks` keeps the pinned id so that starting an interval
        from the tray attaches what you pinned this morning.
        """
        return self._request("GET", "/api/tasks")

    def refresh_tasks(self) -> dict:
        return self._request("POST", "/api/tasks/refresh", {})

    def blocker_policy(self) -> dict:
        return self._request("GET", "/api/blocker/policy")

    def report_block(self, kind: str, target: str, action: str = "blocked", details: str | None = None) -> dict:
        return self._request(
            "POST",
            "/api/blocker/events",
            {"kind": kind, "target": target, "action": action, "details": details},
        )

    def send_report(self) -> dict:
        return self._request("POST", "/api/reports/daily/send", {"force": True})

    def patch_settings(self, patch: dict) -> dict:
        return self._request("PUT", "/api/settings", patch)


def _detail(exc: urllib.error.HTTPError) -> str:
    try:
        body = json.loads(exc.read())
    except Exception:
        return f"{exc.code} {exc.reason}"
    detail = body.get("detail") if isinstance(body, dict) else None
    if isinstance(detail, str):
        return detail
    if isinstance(detail, list):
        return "; ".join(
            f"{'.'.join(str(part) for part in item.get('loc', [])[1:])}: {item.get('msg', '')}"
            for item in detail
        )
    return f"{exc.code} {exc.reason}"