Swarmauri RabbitMQ Publisher

RabbitMQPublisher is the Swarmauri PublishBase implementation for RabbitMQ. The publisher opens a pika.BlockingConnection, ensures the target exchange exists, and emits JSON payloads using persistent delivery semantics.

Highlights

• Configure using either a full AMQP URI or discrete host/port credential fields. Supplying both styles raises ValueError.
• Declares the target exchange as a durable direct exchange during instantiation.
• Serializes payloads with json.dumps and publishes with delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE so messages survive broker restarts.
• On pika.exceptions.AMQPConnectionError or ChannelClosedByBroker the publisher reconnects once, re-declares the exchange, and retries the publish.

Installation

Choose the tool that fits your workflow:

# pip
pip install swarmauri_publisher_rabbitmq

# Poetry
poetry add swarmauri_publisher_rabbitmq

# uv
uv add swarmauri_publisher_rabbitmq

Configuration

RabbitMQPublisher extends PublishBase and supports the following fields:

The publish(channel, payload) method treats channel as the RabbitMQ routing key. payload must be JSON serialisable because it is serialized with json.dumps before publishing.

Quickstart

Ensure RabbitMQ is reachable and that you have permission to declare the exchange. The example below shows the host/port configuration path.

# README Quickstart Example
from swarmauri_publisher_rabbitmq import RabbitMQPublisher

publisher = RabbitMQPublisher(
    host="localhost",
    port=5672,
    username="guest",
    password="guest",
    exchange="demo_exchange",
)

publisher.publish(
    channel="demo.routing.key",
    payload={"message": "Hello RabbitMQ!"},
)

The constructor builds the AMQP URI from the supplied parts, opens a blocking connection, and declares demo_exchange as a durable direct exchange. Each call to publish JSON-encodes the payload and requests persistent delivery so the message survives broker restarts.

Connecting with an AMQP URI

If you already have an AMQP URI, provide it directly and omit the individual connection fields:

from swarmauri_publisher_rabbitmq import RabbitMQPublisher

publisher = RabbitMQPublisher(
    uri="amqp://guest:guest@localhost:5672/",
    exchange="demo_exchange",
)

publisher.publish(
    channel="demo.routing.key",
    payload={"message": "Hello RabbitMQ via URI!"},
)

Behavior notes

• Credentials are URL-encoded via urllib.parse.quote_plus when constructing an AMQP URI from discrete fields, and a username can be supplied without a password if the broker permits it.
• pika.BasicProperties is called with delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE.
• If RabbitMQ closes the channel or the connection drops, the publisher closes the connection, recreates it with the original URI, re-declares the exchange, and retries the publish once.
• When the publisher instance is garbage-collected it closes the open connection if it is still active.

Want to help?

If you want to contribute to swarmauri-sdk, read up on our guidelines for contributing that will help you get started.