zenoh-grpc 1.7.2.dev2026041624

Python bindings for zenoh-plugin-grpc

zenoh-grpc-python

Python bindings for zenoh-plugin-grpc.

Install

From this directory:

maturin develop

Or build a wheel:

maturin build

Or simply pip install:

pip install -e .

Start A Server First

Use either:

cargo run -p zenoh-bridge-grpc

or zenohd with the plugin loaded.

Minimal Example

import zenoh_grpc

with zenoh_grpc.Session.connect() as session:
    pub = session.declare_publisher("demo/example", encoding="text/plain")
    pub.put(b"hello", encoding="text/plain")

Session.connect() defaults to unix:///tmp/zenoh-grpc.sock, so the shortest form is:

with zenoh_grpc.Session.connect() as session:
    ...

Python-Style Optional Parameters

The binding keeps Python-style direct function calls and exposes extra gRPC options as optional keyword arguments. If you omit them, the plugin now falls back to Zenoh's native defaults instead of hard-coded wrapper defaults.

put/delete/reply* calls are enqueue-style: success means the request entered a local bounded queue. Slow receivers are also isolated behind local bounded queues, and when a queue is full the oldest item is dropped. Subscriber.dropped_count(), QueryStream.dropped_count(), ReplyStream.dropped_count(), Publisher.send_dropped_count(), and Queryable.send_dropped_count() expose those counters.

declare_subscriber(key_expr, callback=None, ...) and declare_queryable(key_expr, callback=None, ...) support inline callbacks. When a callback is provided, the binding runs it on a dedicated OS thread for that object and acquires the GIL for each event. Callback mode is exclusive with manual receive APIs. Queryables expose receiver() for pull mode, yielding a QueryStream that produces Query objects, and Session.get() / Querier.get() return iterable ReplyStream objects.

After replying to a query, call query.drop() when you are done, or use with receiver.recv() as query: so the query is finalized automatically on context-manager exit.

import zenoh_grpc

with zenoh_grpc.Session.connect() as session:
    pub = session.declare_publisher(
        "demo/example",
        encoding="text/plain",
        express=True,
    )
    pub.put(b"hello", encoding="text/plain")

    for reply in session.get("demo/example", timeout_ms=3000):
        if reply.ok:
            print(reply.sample.key_expr, reply.sample.payload)

Supported endpoint formats:

• unix:///tmp/zenoh-grpc.sock
• tcp://127.0.0.1:7335

Examples

See:

• examples/pub.py
• examples/sub.py
• examples/sub_callback.py
• examples/queryable.py
• examples/queryable_callback.py
• examples/querier.py
• examples/get.py