kombu review
Kombu 5.6.2 is the messaging layer used by Celery, available directly when an application needs producers and consumers without a task framework. Its `Connection`, `Exchange`, `Queue`, `Producer`, and `Consumer` objects present an AMQP-shaped API over RabbitMQ and a set of virtual transports including Redis and SQS. Serialization, declaration, retry, recovery, and connection pools sit in the same package, but broker features are not equal across transports. The current patch fixes Redis credential-provider compatibility, forwards ACL credentials through Redis Sentinel, improves Pub/Sub acknowledgement-deadline errors, and simplifies requirements around the Python 3.9 floor.
Kombu 5.6.2 installed in 0.3 seconds and used 5 MB across 5 packages in our sandbox, but production still inherits the selected broker's delivery rules. Choose it for explicit AMQP-style messaging across supported transports; choose a broker-native client when portability would only hide important differences.
We installed it
| Install | ✓ · 0.3s | 5 packages on disk · 5 MB |
| Import | ✓ | import kombu in 0.08s · pure Python · requires Python >=3.9 |
| Known vulns | 0 | (pip-audit) |
Answers from our run
Does kombu install cleanly?
Yes. In a fresh container with an empty cache, pip install kombu finished in 0.3s, leaving 5 packages and 5 MB on disk. pip-audit reported no known vulnerabilities.
What does kombu need to run?
Python >=3.9, and nothing compiled: it is pure Python. In our run import kombu succeeded in 0.08s.
kombu or pika: which should you use?
pika: Use it for direct AMQP 0-9-1 access when RabbitMQ is the only target. Kombu 5.6.2 installed in 0.3 seconds and used 5 MB across 5 packages in our sandbox, but production still inherits the selected broker's delivery rules.
When should you not use kombu?
The product needs task decorators, scheduled jobs, task-state persistence, or worker management. Kombu moves messages and leaves that job system to Celery or another framework.
Use it if
- A Python service needs exchange, routing-key, queue, acknowledgement, and retry control without Celery's task registry or result model.
- RabbitMQ is the current broker but a documented virtual transport may be required for another deployment.
- Publishers need connection recovery, channel revival, or a bounded producer pool around repeated sends.
- Unit tests can use the in-memory transport while broker integration tests exercise the actual RabbitMQ, Redis, or SQS behavior.
- The product needs task decorators, scheduled jobs, task-state persistence, or worker management. Kombu moves messages and leaves that job system to Celery or another framework.
- Consumers must run as native asyncio tasks. Kombu's normal `drain_events` loop blocks; aio-pika is designed around asyncio for RabbitMQ.
- Only RabbitMQ will ever be used and direct AMQP details are welcome. Pika removes the virtual-transport layer and exposes broker behavior more plainly.
- Every backend must support the same queue semantics. Kombu's comparison table shows missing priority, message TTL, fanout, and monitoring features on several transports.
- Strict typing requires a PEP 561 marker from every dependency. Our 5.6.2 wheel did not contain `py.typed`.
Setup reality
We installed Kombu 5.6.2 in a clean Python 3.12 Bookworm container. It completed in 0.3 seconds, left 5 packages using 5 MB, and pip-audit found 0 known vulnerabilities. import kombu took 0.08 seconds. The distribution is pure Python, requires Python 3.9 or newer, declares 27 direct dependency entries, uses BSD-3-Clause, and has no py.typed marker. The base import needs no running broker.
A useful connection does need a broker URL and, usually, credentials. Non-AMQP transports require their matching extras, such as Redis or SQS support. Treat the complete URL as a secret because usernames and passwords may be embedded in it. Extras can add platform-sensitive clients; test the chosen transport in the deployment image. Version 5.6.2 specifically repairs Redis providers older than 5.3 and ensures Sentinel master discovery receives ACL credentials.
Declare topology on both ends. A producer can pass declare=[queue], and a consumer declares its queue when entering the consumer context. Several virtual transports keep declarations in client memory rather than in a broker-native exchange model. JSON is the sensible default serializer. Kombu rejects disabled content types, and enabling pickle for messages from untrusted publishers permits Python object construction during decode.
A consumer waits in blocking drain_events and must acknowledge, reject, or requeue each delivery. Connection recovery cannot guarantee one execution: losing the link after the broker accepts a publish or before it records an acknowledgement can produce a duplicate. Handlers need idempotency. Visibility timeouts, fanout, TTL, priority, and monitoring vary by transport. Version 5.6.2 improves Pub/Sub acknowledgement-extension errors, but it does not erase those broker differences.
Patterns
Describe an exchange and bound queue declare-routing
from kombu import Exchange, Queue
events = Exchange("events", type="topic", durable=True)
audit = Queue(
"audit.events",
exchange=events,
routing_key="audit.#",
durable=True,
)These are declarations until bound to a channel. Topic routing and durability depend on the selected transport.
Publish JSON after declaring the route publish-json
from kombu import Connection
with Connection(broker_url) as connection:
producer = connection.Producer(serializer="json")
producer.publish(
{"event": "invoice.created", "id": 42},
exchange=events,
routing_key="audit.invoice",
declare=[audit],
retry=True,
)Publish retry covers recoverable connection errors. A connection loss after broker acceptance can still result in a duplicate send.
Consume JSON and acknowledge success consume-and-ack
def handle(body, message):
process(body)
message.ack()
with Connection(broker_url) as connection:
with connection.Consumer(audit, callbacks=[handle], accept=["json"]):
while True:
connection.drain_events(timeout=5)`drain_events` blocks. An exception before `ack()` may cause redelivery, so `process` must tolerate the same message twice.
Reject a permanently invalid payload reject-invalid-message
def handle(body, message):
if "id" not in body:
message.reject(requeue=False)
return
process(body)
message.ack()With `requeue=False`, the broker discards or dead-letters according to its configuration. Requeue only failures that another attempt may fix.
Exercise message wiring without a broker test-memory-transport
from kombu import Connection, Queue
queue = Queue("test-events")
with Connection("memory://") as connection:
connection.Producer(serializer="json").publish(
{"ok": True}, routing_key=queue.name, declare=[queue]
)
message = queue(connection).get(no_ack=True)
assert message.payload == {"ok": True}The memory transport is process-local and does not reproduce RabbitMQ, Redis, or SQS failure and durability behavior.
Bound startup connection retries retry-initial-connection
from kombu import Connection
connection = Connection(broker_url)
connection.ensure_connection(
max_retries=5,
interval_start=1,
interval_step=2,
interval_max=10,
)A finite `max_retries` lets the process fail health checks instead of hanging forever while the broker is unavailable.
Retry an operation on a revived channel revive-producer
with Connection(broker_url) as connection:
producer = connection.Producer(serializer="json")
publish = connection.ensure(producer, producer.publish, max_retries=3)
publish(
{"id": 42},
exchange=events,
routing_key="audit.created",
declare=[audit],
)`ensure` can replace a broken connection and channel. It cannot tell whether the broker accepted the last attempt before disconnection.
Set Redis visibility and polling options configure-redis-transport
from kombu import Connection
connection = Connection(
"redis://localhost:6379/0",
transport_options={
"visibility_timeout": 3600,
"polling_interval": 2,
"client_name": "billing-consumer",
},
)These options are Redis-specific. Set visibility above the expected processing time or long handlers can be delivered again while still running.
Reuse a producer from the global pool borrow-producer
from kombu import Connection
from kombu.pools import producers
connection = Connection(broker_url)
with producers[connection].acquire(block=True) as producer:
producer.publish(
{"id": 42},
exchange=events,
routing_key="audit.created",
declare=[audit],
serializer="json",
)The context returns the producer after success or failure. Pool identity follows the connection settings used as the key.
Send an already encoded byte payload publish-bytes
with Connection(broker_url) as connection:
connection.Producer().publish(
b"encoded-payload",
exchange=events,
routing_key="audit.raw",
content_type="application/octet-stream",
content_encoding="binary",
declare=[audit],
)When bypassing serializers, set content metadata and make the consumer accept and decode that exact type.
Bind a queue before declaring it bind-and-declare
from kombu import Connection
with Connection(broker_url) as connection:
with connection.channel() as channel:
bound = audit(channel)
bound.declare()Calling broker methods on an unbound `Queue` raises `NotBoundError`; binding returns a channel-specific copy.
Consume two routes on one channel consume-two-queues
with Connection(broker_url) as connection:
with connection.Consumer(
[audit_queue, billing_queue],
callbacks=[handle],
accept=["json"],
prefetch_count=20,
):
while True:
connection.drain_events(timeout=5)The prefetch count applies to deliveries held by this consumer channel; tune it against handler duration and fairness.
Alternatives
More infra guides
boto3 · opentelemetry-api · psutil · distro · @opentelemetry/api · google-cloud-storage · the whole shelf →
How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.