feat(python-sdk): callback subscribers, queryables, and publisher convenience methods

[lferraz]

Summary

Adds callback subscribers, queryables, and publisher convenience methods to NodeContext, completing the Python SDK's API surface for event-driven nodes.

Callback subscribers

Event-driven pattern — no polling loop needed. The handler is called each time a message arrives, auto-decoded via the existing SchemaRegistry (proto, JSON, or raw bytes). Pass max_workers to offload slow handlers to a thread pool:

# Fast handler — runs on Zenoh's thread (default)
sub = ctx.subscriber_callback("sensor/data", on_msg)

# Slow handler (DB, HTTP) — runs in a thread pool
sub = ctx.subscriber_callback("sensor/data", on_msg, max_workers=4)

# Raw variant — handler receives zenoh.Sample directly
sub = ctx.subscriber_raw_callback("bubbaloop/**/health", on_health)

Queryables

Respond to Zenoh GET requests. Same max_workers pattern:

qbl = ctx.queryable("status", on_query)
qbl = ctx.queryable("status", on_query, max_workers=4)  # slow handler
qbl = ctx.queryable_raw("bubbaloop/**/schema", on_schema)  # literal key expr

Publisher convenience methods

NodeContext now exposes publisher_json(), publisher_proto(), and publisher_raw() for declaring publishers with the correct encoding.

New classes

Class	Description
CallbackSubscriber	Event-driven, auto-decode, optional thread pool
RawCallbackSubscriber	Event-driven, raw zenoh.Sample, optional thread pool
Queryable	Queryable handler, optional thread pool

All inherit _BaseSubscriber (except Queryable) with idempotent undeclare().

New NodeContext methods

Method	Description
publisher_json(suffix)	JSON publisher
publisher_proto(suffix, msg_class)	Protobuf publisher
publisher_raw(suffix, local=False)	Raw bytes publisher
subscriber_callback(suffix, handler, max_workers=None)	Callback with auto-decode
subscriber_raw_callback(key_expr, handler, max_workers=None)	Callback with raw Sample
queryable(suffix, handler, max_workers=None)	Queryable at topic prefix
queryable_raw(key_expr, handler, max_workers=None)	Queryable at literal key expr

Other changes

• Added CLAUDE.md and CONTRIBUTING.md with SDK development conventions
• Added pixi.toml + pixi.lock for reproducible dev environment
• Robust run_node() shutdown with heartbeat join and try/finally
• 68 unit tests (from 23 in upstream), all passing deterministically

Test plan

• 68 unit tests passing (pixi run python -m pytest tests/ -v -p no:randomly)
• Lint clean (pixi run check)
• All public imports verified

🤖 Generated with Claude Code

[qodo-code-review]

Review Summary by Qodo

Add callback subscribers, async queryables, and queue-backed blocking subscribers with thread pool support

✨ Enhancement 🧪 Tests

Walkthroughs

Description

• Add callback-based and async subscribers with thread pool support
• Implement queryable factory methods on NodeContext with async variants
• Convert blocking subscribers to queue-backed with optional timeout
• Export new subscriber/queryable classes and add comprehensive test coverage
• Expand project metadata, linting config, and developer documentation

Diagram

flowchart LR
  A["Blocking Subscribers<br/>TypedSubscriber/RawSubscriber"] -->|"queue-backed<br/>with timeout"| B["recv timeout support"]
  C["Callback Subscribers<br/>CallbackSubscriber/RawCallbackSubscriber"] -->|"Zenoh thread"| D["Event-driven handlers"]
  E["Async Subscribers<br/>CallbackSubscriberAsync/RawCallbackSubscriberAsync"] -->|"ThreadPoolExecutor"| F["Non-blocking handlers"]
  G["Queryables<br/>queryable/queryable_raw"] -->|"Zenoh thread"| H["Query handlers"]
  I["Async Queryables<br/>AsyncQueryable"] -->|"ThreadPoolExecutor"| J["Slow query handlers"]
  K["NodeContext"] -->|"factory methods"| A
  K -->|"factory methods"| C
  K -->|"factory methods"| E
  K -->|"factory methods"| G
  K -->|"factory methods"| I

Loading

File Changes

1. python-sdk/bubbaloop_sdk/subscriber.py ✨ Enhancement +224/-23

Implement callback and async subscriber variants with thread pools

python-sdk/bubbaloop_sdk/subscriber.py

---

2. python-sdk/bubbaloop_sdk/context.py ✨ Enhancement +123/-5

Add factory methods for callback/async subscribers and queryables

python-sdk/bubbaloop_sdk/context.py

---

3. python-sdk/bubbaloop_sdk/__init__.py ✨ Enhancement +15/-2

Export new subscriber and queryable classes in public API

python-sdk/bubbaloop_sdk/init.py

---

View more (17)

4. python-sdk/tests/test_context.py 🧪 Tests +681/-3

Add 48 unit tests for all subscriber and queryable variants

python-sdk/tests/test_context.py

---

5. python-sdk/pyproject.toml ⚙️ Configuration changes +90/-1

Expand metadata, add ruff/pytest/coverage configuration

python-sdk/pyproject.toml

---

6. python-sdk/pixi.toml ⚙️ Configuration changes +22/-0

Add pixi development tasks for test, lint, format, check

python-sdk/pixi.toml

---

7. python-sdk/README.md 📝 Documentation +142/-76

Rewrite with synchronous API examples and comprehensive reference

python-sdk/README.md

---

8. python-sdk/CONTRIBUTING.md 📝 Documentation +81/-0

Add contributor guide with pixi tasks and lint configuration

python-sdk/CONTRIBUTING.md

---

9. python-sdk/CLAUDE.md 📝 Documentation +110/-0

Add conventions, threading model, and pitfalls documentation

python-sdk/CLAUDE.md

---

10. python-sdk/get_sample.py 🐞 Bug fix +0/-0

Fix exception chaining with B904 raise-from pattern

python-sdk/get_sample.py

---

11. python-sdk/decode_sample.py Formatting +0/-0

Reorganize imports for clarity and consistency

python-sdk/decode_sample.py

---

12. python-sdk/health.py Miscellaneous +0/-0

Remove unused time import

python-sdk/health.py

---

13. python-sdk/node.py Miscellaneous +0/-0

Remove unused imports

python-sdk/node.py

---

14. python-sdk/context.py ✨ Enhancement +0/-0

Add TYPE_CHECKING guard for cross-module type annotations

python-sdk/context.py

---

15. .github/workflows/ci.yml ⚙️ Configuration changes +11/-0

Add Python SDK lint and test steps to CI pipeline

.github/workflows/ci.yml

---

16. .markdownlint.json ⚙️ Configuration changes +4/-0

Add markdown linting configuration at repository root

.markdownlint.json

---

17. python-sdk/bubbaloop_sdk/decode_sample.py Additional files +2/-1

...

python-sdk/bubbaloop_sdk/decode_sample.py

---

18. python-sdk/bubbaloop_sdk/get_sample.py Additional files +2/-2

...

python-sdk/bubbaloop_sdk/get_sample.py

---

19. python-sdk/bubbaloop_sdk/health.py Additional files +0/-1

...

python-sdk/bubbaloop_sdk/health.py

---

20. python-sdk/bubbaloop_sdk/node.py Additional files +0/-2

...

python-sdk/bubbaloop_sdk/node.py

---

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (3) 📘 Rule violations (1) 📎 Requirement gaps (0) 🎨 UX Issues (0)

1. CI runs ruff/pytest directly 📘 Rule violation ⛨ Security

Description

The new CI steps execute ruff and pytest as top-level commands, which are not in the allowlisted
command prefixes. This violates the build-command execution restriction and could broaden the set of
runnable commands beyond the approved tooling prefixes.

Code

.github/workflows/ci.yml[R53-62]

+      - name: Python SDK — lint
+        run: |
+          cd python-sdk
+          pip install -e ".[dev]" -q
+          ruff check bubbaloop_sdk/ tests/
+
+      - name: Python SDK — test
+        run: |
+          cd python-sdk
+          pytest tests/ -v

Evidence

PR Compliance ID 10 restricts executed build commands to allowlisted prefixes (cargo, pixi,
npm, make, python, pip). The added workflow steps invoke ruff and pytest directly (not
via python -m ... or pixi run ...).

CLAUDE.md
.github/workflows/ci.yml[53-62]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The CI workflow runs `ruff` and `pytest` directly, which is outside the allowlisted command prefixes required by compliance.

## Issue Context
Compliance requires executed build/test commands to use one of these prefixes only: `cargo`, `pixi`, `npm`, `make`, `python`, `pip`. In CI, `ruff`/`pytest` should be invoked through `python -m ruff ...` / `python -m pytest ...` (or via `pixi run ...`) so the executed command prefix is allowlisted.

## Fix Focus Areas
- .github/workflows/ci.yml[53-62]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Submit after shutdown 🐞 Bug ☼ Reliability

Description

CallbackSubscriberAsync/RawCallbackSubscriberAsync/AsyncQueryable can call
ThreadPoolExecutor.submit() from a Zenoh callback while undeclare() is concurrently shutting the
executor down, raising RuntimeError("cannot schedule new futures after shutdown") in Zenoh’s
callback thread. This can break callback delivery during shutdown and potentially affect other
callbacks/queryables sharing the same Zenoh session thread.

Code

python-sdk/bubbaloop_sdk/subscriber.py[R179-196]

+    def __init__(self, session: zenoh.Session, topic: str, handler, msg_class=None, max_workers: int = 4):
+        self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
+
+        def _wrap(sample: zenoh.Sample) -> None:
+            payload = bytes(sample.payload.to_bytes())
+            if msg_class is not None and hasattr(msg_class, "FromString"):
+                msg = msg_class.FromString(payload)
+            else:
+                msg = payload
+            self._executor.submit(handler, msg)
+
+        self._sub = session.declare_subscriber(topic, _wrap)
+
+    def undeclare(self) -> None:
+        """Undeclare the subscriber and shutdown the thread pool."""
+        self._sub.undeclare()  # stop Zenoh callbacks first
+        self._executor.shutdown(wait=False)
+

Evidence

The Zenoh callback wrapper submits work to the executor, while undeclare() shuts the executor down;
submit() after shutdown deterministically raises RuntimeError in Python, and there is no
guard/try-except to prevent this if a callback is in-flight during undeclare(). The same pattern
exists in all three async variants.

python-sdk/bubbaloop_sdk/subscriber.py[179-196]
python-sdk/bubbaloop_sdk/subscriber.py[207-219]
python-sdk/bubbaloop_sdk/subscriber.py[244-255]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`CallbackSubscriberAsync`, `RawCallbackSubscriberAsync`, and `AsyncQueryable` submit to a `ThreadPoolExecutor` from Zenoh callbacks while `undeclare()` can concurrently call `executor.shutdown()`. If an in-flight callback races with shutdown, `executor.submit(...)` raises `RuntimeError: cannot schedule new futures after shutdown`, potentially surfacing on Zenoh’s internal thread.

### Issue Context
Even though `undeclare()` calls `_sub.undeclare()`/`_qbl.undeclare()` first, this does not guarantee no callback is currently executing; an in-flight callback can still attempt to submit after shutdown begins.

### Fix Focus Areas
- python-sdk/bubbaloop_sdk/subscriber.py[179-196]
- python-sdk/bubbaloop_sdk/subscriber.py[207-219]
- python-sdk/bubbaloop_sdk/subscriber.py[244-255]

### Suggested approach
- Add a `_closing` flag (e.g., `threading.Event` or boolean protected by a lock).
- In `_wrap`, return early if closing.
- Wrap `self._executor.submit(...)` in `try/except RuntimeError` and drop/log if shutting down.
- Consider `shutdown(cancel_futures=True)` (Python 3.9+) so queued-but-not-started tasks don’t keep the process alive during shutdown.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Blocking recv can hang 🐞 Bug ☼ Reliability

Description

TypedSubscriber/RawSubscriber block on queue.get(timeout=None) but undeclare() only undeclares
the Zenoh subscriber and does not wake any thread currently blocked in recv()/iteration. If a
caller uses recv() without a timeout (or iterates) and then tries to stop via undeclare(), that
thread can remain blocked indefinitely.

Code

python-sdk/bubbaloop_sdk/subscriber.py[R38-56]

+    def recv(self, timeout: float | None = None):
+        """Block until the next message arrives. Returns ``None`` on timeout."""
+        try:
+            return self._queue.get(timeout=timeout)
+        except queue.Empty:
+            return None

    def __iter__(self):
        return self

    def __next__(self):
-        try:
-            return self.recv()
-        except Exception as exc:
-            raise StopIteration from exc
+        msg = self.recv()
+        if msg is None:
+            raise StopIteration
+        return msg

    def undeclare(self) -> None:
+        """Undeclare the subscriber and stop receiving samples."""
        self._sub.undeclare()

Evidence

Both subscribers use queue.Queue().get(...) as the blocking primitive; undeclare() does not
enqueue a sentinel or otherwise signal the queue, so a blocked consumer has no wake-up path once no
further samples arrive after undeclare().

python-sdk/bubbaloop_sdk/subscriber.py[25-57]
python-sdk/bubbaloop_sdk/subscriber.py[66-93]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`TypedSubscriber.recv()` / `RawSubscriber.recv()` block on `queue.get(timeout=None)` and `undeclare()` doesn’t wake a blocked consumer. This can leave threads stuck during shutdown if users call `recv()` without a timeout or rely on iteration.

### Issue Context
The PR introduces `recv(timeout=...)` for shutdown-aware loops, but the classes still expose a blocking mode (`timeout=None`) and `__next__` uses that mode.

### Fix Focus Areas
- python-sdk/bubbaloop_sdk/subscriber.py[38-57]
- python-sdk/bubbaloop_sdk/subscriber.py[74-93]

### Suggested approach
- Add a private sentinel object (e.g., `_CLOSED = object()`).
- In `undeclare()`, set a `_closed` flag and `put(_CLOSED)` (or `put_nowait`) to unblock one waiter; optionally drain/wake multiple waiters.
- In `recv()`, if sentinel received, return `None` (or raise `StopIteration` in `__next__`).
- Document behavior clearly (what `recv()` returns after undeclare).

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Unbounded backlog growth 🐞 Bug ➹ Performance

Description

Queue-backed subscribers and async callback/queryable variants have no backpressure: queue.Queue()
is unbounded and ThreadPoolExecutor’s work queue is unbounded, so sustained
producer>consumer/handler throughput can cause unbounded memory growth (and eventual OOM).

Code

python-sdk/bubbaloop_sdk/subscriber.py[R25-43]

    def __init__(self, session: zenoh.Session, topic: str, msg_class=None):
-        self._sub = session.declare_subscriber(topic)
+        self._queue: queue.Queue = queue.Queue()
        self._msg_class = msg_class

-    def recv(self):
-        """Block until the next sample arrives and return the decoded message."""
-        sample = self._sub.recv()
-        payload = bytes(sample.payload.to_bytes())
-        if self._msg_class is not None and hasattr(self._msg_class, "FromString"):
-            return self._msg_class.FromString(payload)
-        return payload
+        def _on_sample(sample: zenoh.Sample) -> None:
+            payload = bytes(sample.payload.to_bytes())
+            if self._msg_class is not None and hasattr(self._msg_class, "FromString"):
+                self._queue.put(self._msg_class.FromString(payload))
+            else:
+                self._queue.put(payload)
+
+        self._sub = session.declare_subscriber(topic, _on_sample)
+
+    def recv(self, timeout: float | None = None):
+        """Block until the next message arrives. Returns ``None`` on timeout."""
+        try:
+            return self._queue.get(timeout=timeout)
+        except queue.Empty:
+            return None

Evidence

Both TypedSubscriber and RawSubscriber allocate a default queue.Queue() (unbounded) and
continuously put(...) into it from the Zenoh callback. The async variants submit every
callback/query to an executor with no admission control, which similarly queues unlimited pending
work.

python-sdk/bubbaloop_sdk/subscriber.py[25-37]
python-sdk/bubbaloop_sdk/subscriber.py[66-73]
python-sdk/bubbaloop_sdk/subscriber.py[179-191]
python-sdk/bubbaloop_sdk/subscriber.py[244-251]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
Current implementations can buffer unlimited messages/tasks (unbounded `queue.Queue()` and unbounded executor work queues). Under load, this can grow memory until the process is killed.

### Issue Context
These classes are intended for long-running robotics/edge processes, where OOM is a serious reliability failure mode.

### Fix Focus Areas
- python-sdk/bubbaloop_sdk/subscriber.py[25-37]
- python-sdk/bubbaloop_sdk/subscriber.py[66-73]
- python-sdk/bubbaloop_sdk/subscriber.py[179-191]
- python-sdk/bubbaloop_sdk/subscriber.py[244-251]

### Suggested approach
- Consider `queue.Queue(maxsize=N)` with a documented policy (block Zenoh thread, drop oldest, drop newest, etc.).
- For async variants, consider a bounded handoff queue + worker threads, or implement a simple semaphore to cap in-flight futures.
- If dropping is acceptable, expose counters/logging so drops are observable.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

ⓘ The new review experience is currently in Beta. Learn more

[Copilot]

Pull request overview

This PR expands the Python SDK’s synchronous Zenoh integration by adding callback-based subscribers, async (thread-pooled) callback variants, and queryable helpers on NodeContext, plus accompanying docs/tooling/CI updates.

Changes:

• Add queue-backed blocking subscribers with recv(timeout) plus new callback + async-callback subscriber classes and AsyncQueryable.
• Extend NodeContext with subscriber_*callback* and queryable*_async helpers; export new public API from __init__.py.
• Update README/tooling (ruff/pytest-cov/pixi), add contributing/conventions docs, and add Python lint/test steps to CI.

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
python-sdk/bubbaloop_sdk/subscriber.py	Implements queue-backed recv(timeout), callback subscribers, async callback subscribers, and AsyncQueryable.
python-sdk/bubbaloop_sdk/context.py	Adds TYPE_CHECKING imports and new NodeContext helper methods for callback subscribers and queryables.
python-sdk/bubbaloop_sdk/init.py	Re-exports newly added public classes via imports and __all__.
python-sdk/tests/test_context.py	Adds/extends tests for new subscriber/queryable variants and public import surface.
python-sdk/README.md	Rewrites docs to reflect synchronous API and documents new callback/queryable APIs.
python-sdk/pyproject.toml	Adds richer project metadata + ruff/pytest/coverage configuration and dev dependencies.
python-sdk/pixi.toml	Adds pixi workspace config and dev tasks (test/lint/fmt/check).
python-sdk/CONTRIBUTING.md	Documents local dev workflow (pixi/venv), linting, and test patterns.
python-sdk/CLAUDE.md	Captures repository conventions and Zenoh/threading pitfalls for contributors.
python-sdk/bubbaloop_sdk/get_sample.py	Fixes exception chaining on timeout (raise ... from err).
python-sdk/bubbaloop_sdk/decode_sample.py	Splits protobuf imports for formatting/lint friendliness.
python-sdk/bubbaloop_sdk/node.py	Removes unused imports.
python-sdk/bubbaloop_sdk/health.py	Removes unused import.
.github/workflows/ci.yml	Adds Python SDK lint + test steps.
.markdownlint.json	Adds markdownlint configuration overrides.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 17 changed files in this pull request and generated 7 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 17 changed files in this pull request and generated 9 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[lferraz]

Fixed — removed the duplicate test_raw_subscriber_undeclare_calls_sub_undeclare.

[lferraz]

Fixed — example now uses ctx.queryable("status", on_query, max_workers=4).

[lferraz]

Fixed — API table now shows queryable(suffix, handler, max_workers=None) and queryable_raw(key_expr, handler, max_workers=None).

[lferraz]

Fixed — updated to reference max_workers pattern and Queryable class name.

[lferraz]

Fixed — annotated as msg_class: type | None = None.

[lferraz]

Fixed — renamed to test_import_queryable.

[lferraz]

Fixed — example now uses print(msg) instead of msg.value.

[lferraz]

Fixed — example now uses print(f"received: {msg}") with a comment explaining the return type varies.

[lferraz]

Fixed — updated to list actual class names: ProtoSubscriber, RawSubscriber, CallbackSubscriber, RawCallbackSubscriber, Queryable.