vacuum-wall/daemon/client.py

"""Sync daemon client for the web UI.

Communicates with vacuum-walld over a Unix socket using requests-unixsocket.
"""

import json
import logging
import urllib.parse
from typing import Any

import requests
import requests_unixsocket

logger = logging.getLogger(__name__)


class NotFound(Exception):
    """Raised when the daemon returns HTTP 404."""

    pass


class BadRequest(Exception):
    """Exception raised when the daemon returns HTTP 400.

    Used to signal client-side input or formatting errors from the
    daemon for distinction from general failures.
    """

    pass


_DEFAULT_SOCKET = None


def _get_socket_path() -> str:
    """Return the daemon Unix socket path.

    Lazily resolves the path from the VACUUM_WALLD_SOCKET environment
    variable or falls back to data/daemon.sock under the project
    directory.  The result is cached in _DEFAULT_SOCKET.

    Returns:
        Absolute path string for the daemon socket.
    """
    global _DEFAULT_SOCKET
    if _DEFAULT_SOCKET is None:
        import os
        from pathlib import Path

        project_dir = Path(__file__).resolve().parent.parent
        socket_env = os.environ.get(
            "VACUUM_WALLD_SOCKET", str(project_dir / "data" / "daemon.sock")
        )
        _DEFAULT_SOCKET = socket_env
    return _DEFAULT_SOCKET


def set_socket_path(path: str) -> None:
    """Override the default daemon socket path.

    Useful for tests that need an alternate socket.

    Args:
        path: Absolute path to the Unix socket file.
    """
    global _DEFAULT_SOCKET
    _DEFAULT_SOCKET = path


def request(
    method: str,
    path: str,
    json_body: dict[str, Any] | None = None,
    query_params: dict[str, Any] | None = None,
    socket_path: str | None = None,
    timeout: float = 30,
) -> dict[str, Any]:
    """Make a request to the daemon and return the parsed response body.

    For GET requests, query_params are sent as URL query parameters instead
    of a JSON body. For other methods, json_body is sent as JSON.

    Raises RuntimeError on non-2xx responses or connection errors.
    Raises NotFound on HTTP 404. Raises BadRequest on HTTP 400.
    """
    sp = socket_path or _get_socket_path()
    url = f"http+unix://{urllib.parse.quote(sp, safe='')}{path}"
    sess = requests_unixsocket.Session()
    try:
        kwargs: dict[str, Any] = {
            "timeout": timeout,
        }
        if method == "GET":
            if query_params:
                kwargs["params"] = query_params
        else:
            if json_body is not None:
                kwargs["json"] = json_body
        resp = sess.request(
            method,
            url,
            **kwargs,
        )
        resp.raise_for_status()
        try:
            data = resp.json()
        except json.JSONDecodeError as exc:
            raise RuntimeError(
                f"Daemon returned non-JSON response ({resp.status_code}): {exc}"
            ) from exc
    except requests.ConnectionError as exc:
        raise RuntimeError(f"Cannot connect to daemon at {sp}: {exc}") from exc
    except requests.Timeout as exc:
        raise RuntimeError(f"Daemon request timed out: {method} {path}") from exc
    except requests.HTTPError as exc:
        try:
            data = resp.json()
        except Exception:
            data = {"ok": False, "error": resp.text}
        if not data.get("ok"):
            if resp.status_code == 404:
                raise NotFound(data.get("error", str(exc))) from exc
            if resp.status_code == 400:
                raise BadRequest(data.get("error", str(exc))) from exc
            raise RuntimeError(data.get("error", str(exc))) from exc
    if not data.get("ok"):
        raise RuntimeError(data.get("error", "Unknown error"))
    return data.get("data")


def get(path: str, params: dict[str, Any] | None = None, **kwargs: Any) -> Any:
    """Send a GET request to the daemon.

    Query parameters are passed as URL params rather than a JSON body.
    Additional keyword arguments are forwarded to request().

    Args:
        path: URL path to request on the daemon.
        params: Optional query parameters to append to the URL.
        **kwargs: Extra arguments forwarded to request().

    Returns:
        The parsed JSON response data from the daemon.
    """
    return request("GET", path, query_params=params, **kwargs)


def post(path: str, body: dict[str, Any] | None = None, **kwargs: Any) -> Any:
    """Send a POST request to the daemon.

    The body is transmitted as a JSON payload. Extra keyword arguments
    are forwarded to request().

    Args:
        path: URL path to request on the daemon.
        body: Optional JSON-serializable payload.
        **kwargs: Extra arguments forwarded to request().

    Returns:
        The parsed JSON response data from the daemon.
    """
    return request("POST", path, json_body=body, **kwargs)


def patch(path: str, body: dict[str, Any] | None = None, **kwargs: Any) -> Any:
    """Send a PATCH request to the daemon.

    The body is transmitted as a JSON payload. Extra keyword arguments
    are forwarded to request().

    Args:
        path: URL path to request on the daemon.
        body: Optional JSON-serializable payload.
        **kwargs: Extra arguments forwarded to request().

    Returns:
        The parsed JSON response data from the daemon.
    """
    return request("PATCH", path, json_body=body, **kwargs)


def delete(path: str, body: dict[str, Any] | None = None, **kwargs: Any) -> Any:
    """Send a DELETE request to the daemon.

    The body is transmitted as a JSON payload. Extra keyword arguments
    are forwarded to request().

    Args:
        path: URL path to request on the daemon.
        body: Optional JSON-serializable payload.
        **kwargs: Extra arguments forwarded to request().

    Returns:
        The parsed JSON response data from the daemon.
    """
    return request("DELETE", path, json_body=body, **kwargs)


def batch(ops: list[dict[str, Any]], **kwargs: Any) -> dict[str, Any]:
    """Batch request to daemon.

    Each op is a dict with 'id', 'method', 'path', and optionally 'body'.
    Returns a dict mapping each id to its result.
    """
    return post("/batch", {"ops": ops}, **kwargs)