mq-bridge-py 0.2.21

Python bindings for mq-bridge (full: all brokers incl. Kafka)

mq-bridge Python bindings

Thin Python bindings for the Rust mq-bridge core.

Install

Pick exactly one distribution. Both install the same import path: mq_bridge.

Package	Install	Includes
Full	pip install mq-bridge-py	Basic set plus Kafka, AWS, gRPC, MongoDB, SQLx
Basic	pip install mq-bridge-py-basic	HTTP, NATS, MQTT, AMQP, WebSocket, ZeroMQ, middleware

Memory and file endpoints are always present in both packages. Use mq-bridge-py-basic when you want the lean all-platform wheel set. Use mq-bridge-py when you need Kafka or the heavier non-messaging backends.

The public API stays close to mq-bridge itself:

• Route.from_yaml(path, name) loads one named route from a YAML file
• Route.from_yaml_str(text, name) / Route.from_config(mapping, name) build a route from an in-memory YAML string or a Python dict, no file required
• Route.with_handler(...) attaches a raw Message handler, with lazy json()/text() readers and with_json()/with_payload() response helpers
• Route.add_handler(kind, ...) uses mq-bridge's kind dispatch and delivers decoded JSON
• RetryableError and NonRetryableError let Python handlers signal retry intent
• Publisher.from_yaml(path, name) (plus from_yaml_str / from_config) loads one named publisher
• Publisher.send_json(...) and Publisher.request_json(...) serialize Python JSON values in Rust

The Python surface is synchronous and blocking. Tokio, broker I/O, routing, and batching all stay in Rust.

Config types and schema

mq-bridge-app can create and test route and endpoint JSON/YAML through its UI. It does not replace your Python code or handlers, but it is useful when you want a known-good connection and route shape before pasting the configuration into Python. Load the generated config with Route.from_config, Route.from_yaml, Publisher.from_config, or Publisher.from_yaml.

For the from_config / from_yaml_str mappings, mq_bridge.config ships TypedDict definitions so editors autocomplete the config keys (input, output, batch_size, every transport config, middleware, …):

from mq_bridge import Route
from mq_bridge.config import ConfigDocument

config: ConfigDocument = {
    "routes": {
        "orders": {
            "input": {"memory": {"topic": "orders.in", "capacity": 1600}},
            "output": {"response": {}},
            "batch_size": 128,
        }
    }
}
route = Route.from_config(config, "orders")

These types are generated from the JSON Schema, which the extension produces on demand from the Rust models — there is no checked-in schema copy to drift:

from mq_bridge import config_schema

schema = config_schema()        # the JSON Schema as a dict

config_schema() is handy for editor validation of YAML configs too — dump it to a file and point your # yaml-language-server: $schema= line at it. The types are regenerated with uv run python scripts/gen_config_types.py (a test fails if they drift from the schema).

Running a route

Route.run() blocks the calling thread until another thread calls stop() — it deploys the route and then parks. This is convenient for a process whose only job is the route, but it is a common trap: nothing after route.run() executes until the route stops.

To keep running Python code after the route is up, use start() (non-blocking) or the context-manager form:

route = Route.from_config(config, "orders_route").with_handler(handle)

# Non-blocking: deploys, returns, and runs on a background thread.
route.start()
publisher.send_json({"order_id": 42}, {"kind": "order.created"})
route.stop()
route.join()   # optional: wait for a clean shutdown

# Or scope it to a block — starts on enter, stops + joins on exit:
with Route.from_config(config, "orders_route").with_handler(handle):
    publisher.send_json({"order_id": 42}, {"kind": "order.created"})

Configuration/connection errors surface from start() itself, not from a background thread. run() remains available for the blocking single-route case.

Tuning (environment variables)

These knobs are read from the environment at startup:

Variable	Default	Effect
MQ_BRIDGE_PY_HANDLER_EXECUTOR	worker	worker runs handlers on a dedicated interpreter thread that coalesces queued batches under one GIL acquisition (best under load); direct calls the handler inline.
MQ_BRIDGE_PY_HANDLER_CONCURRENCY	CPU count	Max in-flight handler batches. 0 disables the limit.
MQ_BRIDGE_PY_GC_MODE	default	default leaves CPython's cyclic GC alone; count disables it and runs gc.collect() every N messages; off disables it entirely (pure refcounting).
MQ_BRIDGE_PY_GC_THRESHOLD	100000	Messages between collections when MQ_BRIDGE_PY_GC_MODE=count.

Local development

uv is a good fit here for the Python-side developer workflow, while maturin stays the build backend:

cd python/mq-bridge-py
uv sync --group dev --no-install-project
uv run maturin develop
uv run pytest -q

Performance smoke tests are skipped by default because they start routes and measure local throughput:

cd python/mq-bridge-py
MQ_BRIDGE_RUN_PERF_TESTS=1 uv run pytest -q -m performance

Examples

Raw message handler:

cd python/mq-bridge-py
uv run python examples/raw_route.py

Kind-based JSON handler:

cd python/mq-bridge-py
uv run python examples/json_route.py

Memory benchmark:

cd python/mq-bridge-py
uv run maturin develop --release
uv run python examples/bench_memory.py --messages 100000

Analysis

HTTP comparison benchmark, driven by a native load generator (wrk) so the client is never the bottleneck. It boots each server itself (mq-bridge in worker and direct executor modes, plus FastAPI, Starlette, Sanic, aiohttp, and FastStream when installed) and drives each with wrk:

cd python/mq-bridge-py
uv run maturin develop --release
uv sync --group bench   # optional Python HTTP peers
uv run python analysis/bench_http_native.py --connections 1,8,32 --duration 8

Requires wrk on PATH (brew install wrk). The FastStream target compares its ASGI custom-route path over Uvicorn; it is not a broker-backed subscriber/publisher benchmark. The examples use included sample configs or create temporary configs.