From e7b202860c6eb7eddc813807cf3983d93ab43c94 Mon Sep 17 00:00:00 2001 From: Prakriti Gupta Date: Tue, 3 Nov 2026 16:53:27 -0800 Subject: [PATCH] Remove libby argument in LibbyDaemon --- libby/daemon.py | 22 +++++++++++++--------- libby/how_to_create_a_daemon.md | 28 ++++++++++++++-------------- libby/keygrabber/daemon.py | 7 +++---- peers/peer_a.py | 10 +++++----- peers/peer_a_rabbitmq.py | 6 +++--- peers/peer_b.py | 2 +- peers/peer_c.py | 2 +- peers/peer_d.py | 10 +++++----- peers/peer_group.py | 2 +- tests/test_client_integration.py | 4 ++-- tests/test_keygrabber_daemon.py | 2 +- 11 files changed, 49 insertions(+), 46 deletions(-) diff --git a/libby/daemon.py b/libby/daemon.py index 4f11a44..6ec1cb3 100644 --- a/libby/daemon.py +++ b/libby/daemon.py @@ -250,14 +250,18 @@ def _ensure_last_error_handler(self) -> None: ### Optional hooks - def on_start(self, libby: Libby) -> None: - """Initialize hardware and register keywords.""" + def on_start(self) -> None: + """Initialize hardware and register keywords. ``self.libby`` is set.""" - def on_stop(self, libby: Optional[Libby] = None) -> None: - """Release hardware resources.""" + def on_stop(self) -> None: + """Release hardware resources. - def on_hello(self, libby: Libby) -> None: - """Run after the initial discovery hello.""" + Also runs after a start that failed partway, so ``self.libby`` may be + None; it is closed only after this returns. + """ + + def on_hello(self) -> None: + """Run after the initial discovery hello. ``self.libby`` is set.""" def on_event(self, topic: str, msg: Any) -> None: self.logger.info("%s: %s", topic, msg.env.payload) @@ -418,12 +422,12 @@ def start(self) -> None: if self.config_discovery_enabled(): try: self.libby.hello() - self.on_hello(self.libby) + self.on_hello() except Exception: self.logger.exception("discovery hello failed") try: - self.on_start(self.libby) + self.on_start() except Exception: self.logger.exception("daemon initialization failed") if self.fail_fast_on_start: @@ -454,7 +458,7 @@ def stop(self) -> None: return try: - self.on_stop(self.libby) + self.on_stop() except Exception: self.logger.exception("daemon shutdown hook failed") finally: diff --git a/libby/how_to_create_a_daemon.md b/libby/how_to_create_a_daemon.md index 9dc05a7..82a2203 100644 --- a/libby/how_to_create_a_daemon.md +++ b/libby/how_to_create_a_daemon.md @@ -74,9 +74,9 @@ class MyPeer(LibbyDaemon): "alerts.status": lambda payload: print("[X] status:", payload), } - def on_start(self, libby): + def on_start(self): # Manually teach peers about each other (discovery workaround) - libby.learn_peer_keys("peer-Y", ["service1", "service2"]) + self.libby.learn_peer_keys("peer-Y", ["service1", "service2"]) if __name__ == "__main__": MyPeer().serve() @@ -112,9 +112,9 @@ class MyPeer(LibbyDaemon): "alerts.status": lambda payload: print("[X] status:", payload), } - def on_start(self, libby): + def on_start(self): # Manually teach peers about each other (discovery workaround) - libby.learn_peer_keys("peer-Y", ["service1", "service2"]) + self.libby.learn_peer_keys("peer-Y", ["service1", "service2"]) if __name__ == "__main__": MyPeer().serve() @@ -196,18 +196,18 @@ class PeerA(LibbyDaemon): discovery_enabled = True discovery_interval_s = 2.0 - def on_start(self, libby): + def on_start(self): try: - if not libby.wait_for_key("peer-B", "perf.echo", timeout_s=2.5): - libby.learn_peer_keys("peer-B", ["perf.echo", "ping.txt", "answer"]) + if not self.libby.wait_for_key("peer-B", "perf.echo", timeout_s=2.5): + self.libby.learn_peer_keys("peer-B", ["perf.echo", "ping.txt", "answer"]) except AttributeError: pass print("[PeerA] asking B: perf.echo …") - res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000) + res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000) print("[PeerA] result:", res) - libby.publish("alerts.status", {"source": "peer-A", "ok": True}) + self.libby.publish("alerts.status", {"source": "peer-A", "ok": True}) print("[PeerA] published alerts.status") if __name__ == "__main__": @@ -253,18 +253,18 @@ class PeerC(LibbyDaemon): "math.add": math_add, } - def on_start(self, libby): + def on_start(self): # Proxy service needing live libby handle def echo_proxy(_p: Dict[str, Any]): - res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=6000) + res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=6000) return {"ok": True, "forwarded_to": "peer-B", "result": res} self.add_service("perf.echo.proxy", echo_proxy) print("[PeerC] math.add(2,5) ->", - libby.rpc(self.peer_id, "math.add", {"a": 2, "b": 5}, ttl_ms=2000)) + self.libby.rpc(self.peer_id, "math.add", {"a": 2, "b": 5}, ttl_ms=2000)) - libby.publish("alerts.status", {"source": "peer-C", "ok": True}) + self.libby.publish("alerts.status", {"source": "peer-C", "ok": True}) if __name__ == "__main__": PeerC().serve() @@ -312,5 +312,5 @@ class MyPeer(LibbyDaemon): - **Encapsulation goal**: application peers only implement business logic (`on_req`, `on_event`) and a few config methods. - **Handlers are payload-only.** They receive a Python dict and return anything JSON-serializable. If the return is not a dict, LibbyDaemon auto-wraps it as `{"data": }`. -- **Discovery workaround**: Currently, automatic peer discovery needs manual key learning via `libby.learn_peer_keys()` in `on_start()`. +- **Discovery workaround**: Currently, automatic peer discovery needs manual key learning via `self.libby.learn_peer_keys()` in `on_start()`. diff --git a/libby/keygrabber/daemon.py b/libby/keygrabber/daemon.py index 778a0d7..6a1eb5e 100644 --- a/libby/keygrabber/daemon.py +++ b/libby/keygrabber/daemon.py @@ -13,7 +13,6 @@ from ..config import ConfigError, DaemonConfigLoader, with_env_overrides from ..daemon import LibbyDaemon from ..errors import LibbyError -from ..libby import Libby from .collection import FAILURES_BEFORE_BACKOFF, Collection from .config import TIMEOUT_HEADROOM, KeygrabberConfig, build_sink, parse_config from .scheduler import Scheduler @@ -112,10 +111,10 @@ def __init__(self) -> None: self._reconnect_wanted = False self._clock: Callable[[], float] = time.monotonic - def on_start(self, libby: Libby) -> None: + def on_start(self) -> None: """Parse config, open the sink, and start the reader and writer threads.""" self._settings = parse_config(self._config) - self._client = Client(libby) + self._client = Client(self.libby) self._writer = RetryingWriter(self.make_sink(), policy=self._settings.retry) self._writer.connect() @@ -133,7 +132,7 @@ def on_start(self, libby: Libby) -> None: self.logger.info("collecting %d collections with %d workers", len(self._collections), self._settings.workers) - def on_stop(self, libby: Optional[Libby] = None) -> None: + def on_stop(self) -> None: """Stop the threads and close the sink once the writer has drained it. The draining happens on the writer thread rather than here, so only one diff --git a/peers/peer_a.py b/peers/peer_a.py index 6b40cca..efdbc8d 100644 --- a/peers/peer_a.py +++ b/peers/peer_a.py @@ -17,20 +17,20 @@ class PeerA(LibbyDaemon): discovery_enabled = True discovery_interval_s = 2.0 - def on_start(self, libby): + def on_start(self): try: - if not libby.wait_for_key( + if not self.libby.wait_for_key( "peer-B", "perf.echo", timeout_s=2.5, ): - libby.learn_peer_keys( + self.libby.learn_peer_keys( "peer-B", ["perf.echo", "ping.txt", "answer"], ) print("[PeerA] asking B: perf.echo ...") - result = libby.rpc( + result = self.libby.rpc( "peer-B", "perf.echo", {"t0": time.time()}, @@ -41,7 +41,7 @@ def on_start(self, libby): except Exception as exc: print(f"[PeerA] Peer B request failed: {exc}") - libby.publish( + self.libby.publish( "alerts.status", { "source": self.peer_id, diff --git a/peers/peer_a_rabbitmq.py b/peers/peer_a_rabbitmq.py index 723fc3e..d8e10a1 100644 --- a/peers/peer_a_rabbitmq.py +++ b/peers/peer_a_rabbitmq.py @@ -6,16 +6,16 @@ class PeerA(LibbyDaemon): transport = "rabbitmq" rabbitmq_url = "amqp://localhost" - def on_start(self, libby): + def on_start(self): # Give peer B a moment to register its services time.sleep(1.5) print("[PeerA] asking B: perf.echo …") - res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000) + res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000) print("[PeerA] result:", res) # publish a status - libby.publish("alerts.status", {"source": "peer-A", "ok": True}) + self.libby.publish("alerts.status", {"source": "peer-A", "ok": True}) print("[PeerA] published alerts.status") if __name__ == "__main__": diff --git a/peers/peer_b.py b/peers/peer_b.py index 5be05e3..2c4caf7 100644 --- a/peers/peer_b.py +++ b/peers/peer_b.py @@ -33,7 +33,7 @@ class PeerB(LibbyDaemon): }, } - def on_start(self, libby): + def on_start(self): print("[PeerB] ready") print("[PeerB] services:", list(self.services)) diff --git a/peers/peer_c.py b/peers/peer_c.py index 24186da..aa8328c 100644 --- a/peers/peer_c.py +++ b/peers/peer_c.py @@ -32,7 +32,7 @@ def get_last_status(self, payload): "status": self.last_status, } - def on_start(self, libby): + def on_start(self): print("[PeerC] listening for alerts.status") diff --git a/peers/peer_d.py b/peers/peer_d.py index 81cbe56..5078234 100644 --- a/peers/peer_d.py +++ b/peers/peer_d.py @@ -12,7 +12,7 @@ def ping(self, payload: Payload) -> Dict[str, Any]: def echo(self, payload: Payload) -> Dict[str, Any]: return self.payload(payload) - def on_start(self, libby): + def on_start(self): import time target = "peer-B" key = "perf.echo" @@ -21,17 +21,17 @@ def on_start(self, libby): print(f"[{self.config_peer_id()}] routes: {ab}") # 1) Best-effort - libby.wait_for_peer(target, timeout_s=2.0) + self.libby.wait_for_peer(target, timeout_s=2.0) # 2) If the key isn't known yet (discovery not populated) - if not libby.wait_for_key(target, key, timeout_s=1.0): + if not self.libby.wait_for_key(target, key, timeout_s=1.0): try: - libby.learn_peer_keys(target, [key]) + self.libby.learn_peer_keys(target, [key]) except Exception: pass print(f"[{self.config_peer_id()}] asking {target}: {key} …") - res = libby.rpc(target, key, {"t0": time.time()}, ttl_ms=8000) + res = self.libby.rpc(target, key, {"t0": time.time()}, ttl_ms=8000) print(f"[{self.config_peer_id()}] result: {res}") diff --git a/peers/peer_group.py b/peers/peer_group.py index b81607e..f0ef028 100644 --- a/peers/peer_group.py +++ b/peers/peer_group.py @@ -20,7 +20,7 @@ class GroupTestPeer(LibbyDaemon): "echo": handle_echo, } - def on_start(self, libby): + def on_start(self): print(f"[GroupTestPeer] Started with group_id='{self.group_id}'") print(f"[GroupTestPeer] Expected queue name: libby.group.{self.group_id}.peer.{self.peer_id}") diff --git a/tests/test_client_integration.py b/tests/test_client_integration.py index 33e7662..056c40a 100644 --- a/tests/test_client_integration.py +++ b/tests/test_client_integration.py @@ -52,7 +52,7 @@ class _FixtureDaemon(LibbyDaemon): group_id = GROUP_ID discovery_enabled = False - def on_start(self, libby) -> None: + def on_start(self) -> None: state = {"pos": INITIAL_POSITION} self.keyword_registry.float( "positionvalue", @@ -72,7 +72,7 @@ class _LifecycleDaemon(LibbyDaemon): group_id = GROUP_ID discovery_enabled = False - def on_start(self, libby) -> None: + def on_start(self) -> None: self.keyword_registry.trigger( "shutdown", action=self.request_stop, diff --git a/tests/test_keygrabber_daemon.py b/tests/test_keygrabber_daemon.py index 2f88ea3..b7f9ed7 100644 --- a/tests/test_keygrabber_daemon.py +++ b/tests/test_keygrabber_daemon.py @@ -47,7 +47,7 @@ class _FixtureDaemon(LibbyDaemon): group_id = "hsfei" discovery_enabled = False - def on_start(self, libby) -> None: + def on_start(self) -> None: self.keyword_registry.float("positionvalue", getter=lambda: 7.5, units="mm") self.keyword_registry.bool("isconnected", getter=lambda: True)