beachcomber 0.5.1

A centralized daemon that caches shell state (git, battery, hostname, etc.) so every consumer reads from one fast cache instead of independently forking shells
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

"""Newline-delimited JSON protocol helpers for the beachcomber daemon.

All requests are JSON objects with an ``"op"`` field. All responses are
JSON objects with an ``"ok"`` field that is ``true`` or ``false``.
"""

from __future__ import annotations

import json
from typing import Any, Optional

from .exceptions import ProtocolError, ServerError


def encode_request(op: str, **fields: Any) -> bytes:
    """Serialise a request object to a newline-terminated JSON bytes value.

    Args:
        op: The operation name (``"get"``, ``"poke"``, ``"context"``,
            ``"list"``, or ``"status"``).
        **fields: Additional key/value pairs to include in the request.

    Returns:
        UTF-8 encoded JSON followed by a newline byte.
    """
    payload: dict[str, Any] = {"op": op}
    payload.update(fields)
    return (json.dumps(payload, separators=(",", ":")) + "\n").encode()


def decode_response(line: str) -> dict[str, Any]:
    """Parse a single newline-delimited JSON response line.

    Args:
        line: A single text line received from the daemon.

    Returns:
        Parsed response dict.

    Raises:
        ProtocolError: If the line is not valid JSON or is missing the
            ``"ok"`` field.
        ServerError: If the response contains ``"ok": false``.
    """
    line = line.strip()
    if not line:
        raise ProtocolError("received empty response from daemon")

    try:
        data: dict[str, Any] = json.loads(line)
    except json.JSONDecodeError as exc:
        raise ProtocolError(f"invalid JSON from daemon: {exc}") from exc

    if not isinstance(data, dict):
        raise ProtocolError(f"expected JSON object, got {type(data).__name__}")

    ok = data.get("ok")
    if ok is None:
        raise ProtocolError("response missing 'ok' field")

    if ok is False:
        error_msg = data.get("error", "unknown error")
        raise ServerError(str(error_msg))

    return data


def build_get_request(key: str, path: Optional[str] = None) -> bytes:
    """Build a ``get`` request.

    Args:
        key: Provider key, e.g. ``"git.branch"`` or ``"git"``.
        path: Optional working-directory path for per-directory providers.

    Returns:
        Encoded request bytes.
    """
    kwargs: dict[str, Any] = {"key": key}
    if path is not None:
        kwargs["path"] = path
    return encode_request("get", **kwargs)


def build_poke_request(key: str, path: Optional[str] = None) -> bytes:
    """Build a ``poke`` request.

    Args:
        key: Provider key to recompute.
        path: Optional working-directory path.

    Returns:
        Encoded request bytes.
    """
    kwargs: dict[str, Any] = {"key": key}
    if path is not None:
        kwargs["path"] = path
    return encode_request("poke", **kwargs)


def build_context_request(path: str) -> bytes:
    """Build a ``context`` request to set the default path for a session.

    Args:
        path: Directory path to set as the connection context.

    Returns:
        Encoded request bytes.
    """
    return encode_request("context", path=path)


def build_list_request() -> bytes:
    """Build a ``list`` request."""
    return encode_request("list")


def build_status_request() -> bytes:
    """Build a ``status`` request."""
    return encode_request("status")