Add bulk keyword reads with capability detection, and test both transports

.github/workflows/tests.yml

@@ -0,0 +1,41 @@
+name: Tests
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - "libby/**"
+      - "tests/**"
+      - "pyproject.toml"
+      - "tox.ini"
+      - ".github/workflows/tests.yml"
+  pull_request:
+    branches: [main]
+    paths:
+      - "libby/**"
+      - "tests/**"
+      - "pyproject.toml"
+      - "tox.ini"
+  workflow_dispatch: {}
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install tox
+        run: pip install tox
+
+      # No broker service here on purpose: the RabbitMQ integration cases skip
+      # themselves, and the ZMQ ones cover the same behaviour end to end with
+      # no external dependency.
+      - name: Run tests
+        run: tox

README.md

@@ -38,6 +38,7 @@ python -m unittest discover -s tests
 ```
 
 Most of `tests/` needs no transport at all. `tests/test_client_integration.py`
-is the exception: it starts a real `LibbyDaemon` over RabbitMQ and exercises
-`Client` against it, and skips itself automatically if no broker is reachable
-at `amqp://localhost`.
+is the exception: it starts a real `LibbyDaemon` and exercises `Client`
+against it, once per transport. The ZMQ cases need no external service and
+always run; the RabbitMQ cases skip themselves if no broker is reachable at
+`amqp://localhost`.

docs/source/client.md

@@ -26,6 +26,8 @@ client = Client.zmq(address_book={"hsfei_pickoff": "tcp://host:5555"})
   units, flags); `set(name, value)` → the value the daemon applied.
 - `wait_for(expression, timeout)` → blocks until a keyword satisfies a
   comparison; see below.
+- `list(pattern)` → matching qualified names; `describe(name)` → one keyword's
+  metadata; `read(names)` → many keywords in one request per peer; see below.
 - Failures raise rather than return sentinels: `KeywordError` when the daemon
   rejects a get/set (its message is on `.error`), `LibbyTimeout` when a
   request isn't answered, both subclasses of `LibbyError`. `set` accepts
@@ -70,8 +72,46 @@ to report the value the wait settled on. See {mod}`libby.expression` for the
 accepted expression syntax — currently one comparison between a
 `$`-prefixed keyword and a literal.
 
-Exact names only for now; `%` wildcard reads, `list`, and `describe` are
-planned follow-ons — use the {doc}`CLI <cli>` for those today.
+## Listing, describing and bulk reads
+
+`list` returns fully qualified names, so its result feeds straight back into
+`get`, `show` or `read`:
+
+```python
+names = client.list("hsfei.pickoff.is%")   # ["hsfei.pickoff.isconnected", ...]
+meta = client.describe("hsfei.pickoff.positionvalue")
+meta["type"], meta["units"]                # ("float", "mm")
+```
+
+`read` takes many names and issues one request per peer rather than one per
+keyword, which is what a poller should use:
+
+```python
+values = client.read(names)
+# {"hsfei.pickoff.isconnected": {"ok": True, "value": True}, ...}
+```
+
+Unlike `get` and `set`, `read` never raises for a failed read. Every requested
+name maps to its own response, so one dead peer or one broken getter costs
+only its own entries. Names may span peers; each peer is asked separately.
+Long name lists are split into bounded requests (`chunk_size=`) and merged.
+
+`listing` returns both the matching names and the peer's `services` from a
+single `keys.list` response, for a caller that needs to know whether bulk
+reads are available:
+
+```python
+listing = client.listing("hsfei.pickoff.%")
+if "keys.read" in listing.services:
+    values = client.read(list(listing.names))
+else:                                  # peer on an older libby
+    values = {n: client.show(n) for n in listing.names}
+```
+
+Checking `services` is the only reliable test: `Libby.knows_key` reads the
+discovery registry, which stays empty without discovery, and calling
+`keys.read` to see what happens cannot distinguish an old peer from a dead
+one.
 
 See {mod}`libby.client` in the {doc}`API reference </api/index>` for the
 full method signatures.

docs/source/keywords.md

@@ -47,13 +47,34 @@ client.rpc("my-peer", "position", {"value": 12.5})  # modify
 client.rpc("my-peer", "halt", {"value": 1})         # fire
 ```
 
-Two meta-services are auto-registered on every peer that uses the keyword
+Three meta-services are auto-registered on every peer that uses the keyword
 registry:
 
-- `keys.list` — payload `{"pattern": "..."}` (default `"%"`) → names, sorted.
-  `%` wildcards within a single name.
+- `keys.list` — payload `{"pattern": "..."}` (default `"%"`) → `matches`,
+  names sorted, plus `services` (below). `%` wildcards within a single name.
 - `keys.describe` — payload `{"name": "..."}` → flat metadata dict. Exact
   lookup; no wildcards.
+- `keys.read` — payload `{"names": [...]}` or `{"pattern": "..."}` → `values`,
+  a map of name to that keyword's own show response. Reads a whole peer in one
+  request.
+
+`keys.read` exists because a daemon answers requests one at a time, inline on
+its receive thread. Reading twenty keywords individually does not overlap
+anything on that daemon; it serializes exactly as a batch would, while paying
+twenty dispatch cycles instead of one. Batching matters most for a poller that
+must not crowd out an operator or a control command.
+
+A failing getter is reported inside `values` as
+`{"ok": false, "error": "..."}`, so one broken keyword costs only itself.
+Pattern selection skips write-only keywords, which have nothing to show;
+naming one explicitly still answers with its error.
+
+`keys.list` also reports `services`: the non-keyword keys this peer answers,
+including the `keys.*` meta-services and any RPC service a daemon registered
+itself. It is how a caller tells "this peer has no `keys.read`" from "this
+peer did not answer", because an unknown key is dropped without an ACK and so
+probing for one is indistinguishable from a timeout. A peer running an older
+libby omits the field, which is the negative signal.
 
 `LibbyDaemon` subclasses also get a `lasterror` keyword for free (not just
 any keyword-registry user, since it needs the daemon's own logger): a

libby/__init__.py

@@ -12,7 +12,7 @@
     match_pattern,
 )
 from .keyword_registry import KeywordRegistry
-from .client import Client, WaitResult
+from .client import Client, KeyListing, WaitResult
 from .expression import Comparison, parse_comparison
 from .errors import (
     LibbyError,
@@ -26,6 +26,7 @@
 __all__ = [
     "Libby",
     "Client",
+    "KeyListing",
     "WaitResult",
     "Protocol",
     "MessageBuilder",

libby/cli/libby_cli.py

@@ -13,7 +13,6 @@
 from typing import Any, Dict, List, Optional, Tuple
 from urllib.parse import urlsplit, urlunsplit
 
-from libby.client import DEFAULT_POLL_S, Client, WaitResult
 from libby.config_resolve import (
     DEFAULT_BIND,
     DEFAULT_CONFIG_PATH,
@@ -23,6 +22,7 @@
     resolve_rabbitmq_url,
     resolve_transport,
 )
+from libby.client import DEFAULT_POLL_S, Client, WaitResult
 from libby.errors import KeywordError, LibbyError
 from libby.expression import parse_comparison
 from libby.libby import Libby
@@ -271,7 +271,7 @@ def _emit_wait(expression: str, result: WaitResult, *, as_json: bool) -> int:
     return RC_WAIT_TIMEOUT
 
 
-def _modify_timeout(lib: Libby, peer: str, name: str, user_timeout: Optional[float]) -> float:
+def _modify_timeout(client: Client, qualified: str, user_timeout: Optional[float]) -> float:
     """Resolve the timeout for a modify call.
 
     Precedence: ``--timeout`` flag → ``timeout_s`` from the keyword's
@@ -280,12 +280,11 @@ def _modify_timeout(lib: Libby, peer: str, name: str, user_timeout: Optional[flo
     if user_timeout is not None:
         return user_timeout
     try:
-        resp = _rpc_keys_describe(lib, peer, name, DEFAULT_TIMEOUT_S)
-        if resp.get("ok"):
-            t = resp.get("timeout_s")
-            if t is not None:
-                return float(t)
-    except Exception:
+        described = client.describe(qualified, timeout_s=DEFAULT_TIMEOUT_S).get("timeout_s")
+        if described is not None:
+            return float(described)
+    except (LibbyError, ValueError, TypeError):
+        # Best-effort: an unreadable or non-numeric timeout_s just falls back
         pass
     return DEFAULT_TIMEOUT_S
 
@@ -305,14 +304,6 @@ def _rpc_show_one(lib: Libby, peer: str, name: str, timeout: float) -> Dict[str,
     return _peel(lib.rpc(peer, name, {}, ttl_ms=int(timeout * 1000)))
 
 
-def _rpc_keys_list(lib: Libby, peer: str, pattern: str, timeout: float) -> Dict[str, Any]:
-    return _peel(lib.rpc(peer, "keys.list", {"pattern": pattern}, ttl_ms=int(timeout * 1000)))
-
-
-def _rpc_keys_describe(lib: Libby, peer: str, name: str, timeout: float) -> Dict[str, Any]:
-    return _peel(lib.rpc(peer, "keys.describe", {"name": name}, ttl_ms=int(timeout * 1000)))
-
-
 def cmd_show(namespace: argparse.Namespace) -> int:
     config = load_cli_config(namespace.config)
     group, daemon, keyword = parse_keyword(namespace.keyword, allow_pattern=True)
@@ -324,26 +315,25 @@ def cmd_show(namespace: argparse.Namespace) -> int:
     try:
         lib = _mk_libby(namespace, config)
         if "%" in keyword:
-            list_resp = _rpc_keys_list(lib, peer, keyword, timeout)
-            if not list_resp.get("ok"):
-                return _emit_error(
-                    qualified_arg,
-                    list_resp.get("error", "unknown error"),
-                    as_json=namespace.json,
-                )
-            matches: List[str] = list_resp.get("matches", [])
+            matches = Client(lib).list(qualified_arg, timeout_s=timeout)
             if not matches:
                 return 3
+            # One show per match rather than a single keys.read: the CLI has to
+            # keep working against daemons still running a libby without it,
+            # and a human reading a handful of keywords gains nothing from the
+            # round trip saved
             rows: List[Tuple[str, Dict[str, Any]]] = [
-                (f"{group}.{daemon}.{m}", _rpc_show_one(lib, peer, m, timeout))
-                for m in matches
+                (name, _rpc_show_one(lib, peer, name.rsplit(".", 1)[-1], timeout))
+                for name in matches
             ]
             return _emit_many(rows, as_json=namespace.json)
         return _emit_one(
             qualified_arg,
             _rpc_show_one(lib, peer, keyword, timeout),
             as_json=namespace.json,
         )
+    except LibbyError as ex:
+        return _emit_error(qualified_arg, str(ex), as_json=namespace.json)
     except Exception as ex:
         logger.exception("show %s raised", qualified_arg)
         return _emit_error(qualified_arg, str(ex), as_json=namespace.json)
@@ -357,27 +347,22 @@ def cmd_show(namespace: argparse.Namespace) -> int:
 
 def cmd_list(namespace: argparse.Namespace) -> int:
     config = load_cli_config(namespace.config)
-    group, daemon, pattern = parse_keyword(namespace.pattern, allow_pattern=True)
-    peer = peer_id(group, daemon)
+    # Reject a malformed address here, before opening a transport, so it stays
+    # an argument error; Client.list parses it again for the peer id
+    parse_keyword(namespace.pattern, allow_pattern=True)
     timeout = namespace.timeout if namespace.timeout is not None else DEFAULT_TIMEOUT_S
 
     lib: Optional[Libby] = None
     try:
         lib = _mk_libby(namespace, config)
-        resp = _rpc_keys_list(lib, peer, pattern, timeout)
-        if not resp.get("ok"):
-            return _emit_error(
-                namespace.pattern,
-                resp.get("error", "unknown error"),
-                as_json=namespace.json,
-            )
-        matches: List[str] = resp.get("matches", [])
-        if not matches:
+        qualified_names = Client(lib).list(namespace.pattern, timeout_s=timeout)
+        if not qualified_names:
             if namespace.json:
                 print(json.dumps([], indent=2))
             return 3
-        qualified_names = [f"{group}.{daemon}.{m}" for m in matches]
         return _emit_list(qualified_names, as_json=namespace.json)
+    except LibbyError as ex:
+        return _emit_error(namespace.pattern, str(ex), as_json=namespace.json)
     except Exception as ex:
         logger.exception("list %s raised", namespace.pattern)
         return _emit_error(namespace.pattern, str(ex), as_json=namespace.json)
@@ -392,21 +377,16 @@ def cmd_list(namespace: argparse.Namespace) -> int:
 def cmd_describe(namespace: argparse.Namespace) -> int:
     config = load_cli_config(namespace.config)
     group, daemon, keyword = parse_keyword(namespace.keyword, allow_pattern=False)
-    peer = peer_id(group, daemon)
     qualified = f"{group}.{daemon}.{keyword}"
     timeout = namespace.timeout if namespace.timeout is not None else DEFAULT_TIMEOUT_S
 
     lib: Optional[Libby] = None
     try:
         lib = _mk_libby(namespace, config)
-        resp = _rpc_keys_describe(lib, peer, keyword, timeout)
-        if not resp.get("ok"):
-            return _emit_error(
-                qualified,
-                resp.get("error", "unknown error"),
-                as_json=namespace.json,
-            )
+        resp = Client(lib).describe(qualified, timeout_s=timeout)
         return _emit_describe(qualified, resp, as_json=namespace.json)
+    except LibbyError as ex:
+        return _emit_error(qualified, str(ex), as_json=namespace.json)
     except Exception as ex:
         logger.exception("describe %s raised", qualified)
         return _emit_error(qualified, str(ex), as_json=namespace.json)
@@ -444,7 +424,7 @@ def cmd_modify(namespace: argparse.Namespace) -> int:
     lib: Optional[Libby] = None
     try:
         lib = _mk_libby(namespace, config)
-        timeout = _modify_timeout(lib, peer, keyword, namespace.timeout)
+        timeout = _modify_timeout(Client(lib), qualified, namespace.timeout)
         resp = _peel(lib.rpc(peer, keyword, {"value": value}, ttl_ms=int(timeout * 1000)))
         return _emit_one(qualified, resp, as_json=namespace.json)
     except Exception as ex: