PyRMQ

Python with RabbitMQ—simplified so you won’t have to.

Features

Stop worrying about boilerplating and implementing retry logic on your queues. PyRMQ already does it for you.

• Use out-of-the-box Consumer and Publisher classes created from pika for your projects and tests.

• Custom DLX-DLK-based retry logic for message consumption.

• Message priorities

• Works with Python 3.

• Production ready

Quickstart

PyRMQ is available at PyPI.

$ pip install pyrmq

Just instantiate the feature you want with their respective settings. PyRMQ already works out of the box with RabbitMQ’s default initialization settings.

from pyrmq import Publisher
publisher = Publisher(
    exchange_name="exchange_name",
    queue_name="queue_name",
    routing_key="routing_key",
)
publisher.publish({"pyrmq": "My first message"})

Publish message with priorities

To enable prioritization of messages, instantiate your queue with the queue argument x-max-priority. It takes an integer that sets the number of possible priority values with a higher number commanding more priority. Then, simply publish your message with the priority argument specified. Any number higher than the set max priority is floored or considered the same. Read more about message priorities here

from pyrmq import Publisher
publisher = Publisher(
    exchange_name="exchange_name",
    queue_name="queue_name",
    routing_key="routing_key",
    queue_args={"x-max-priority": 3}
)
publisher.publish({"pyrmq": "My first message"}, priority=1)

Warning

Adding arguments on an existing queue is not possible. If you wish to add queue arguments, you will need to either delete the existing queue then recreate the queue with arguments or simply make a new queue with the arguments.

Consuming

Instantiating a Consumer automatically starts it in its own thread making it non-blocking by default. When run after the code from before, you should be able to receive the published data.

from pyrmq import Consumer


def callback(data):
    print(f"Received {data}!")

consumer = Consumer(
    exchange_name="exchange_name",
    queue_name="queue_name",
    routing_key="routing_key",
)

consumer.start()

DLX-DLK Retry Logic

What if you wanted to retry a failure on a consumed message? PyRMQ offers a custom solution that keeps your message in queues while retrying in an exponential backoff fashion.

This approach uses dead letter exchanges and queues to republish a message to your original queue once it has expired. PyRMQ creates this “retry” queue for you with the default naming convention of appending your original queue with .retry.

from pyrmq import Consumer

def callback(data):
    print(f"Received {data}!")
    raise Exception

consumer = Consumer(
    exchange_name="exchange_name",
    queue_name="queue_name",
    routing_key="routing_key",
    callback=callback,
    is_dlk_retry_enabled=True,
)
consumer.start()

This will start a loop of passing your message between the original queue and the retry queue until it reaches the default number of max_retries.

User Guide

• Installation
  • Stable Version
  • Development Version
    • Git Checkout
    • PyPI
• Usage
  • Publishing
    • Retries
    • Connecting
    • Publishing
    • Max retries reached
  • Publish message with priorities
  • Consuming
    • Retries
    • Connecting
    • DLX-DLK Consumption Retry Logic
    • Max retries reached
• API
  • Publisher Class
  • Consumer Class
• Testing
  • RabbitMQ
    • Run Docker image (recommended)
    • Install and run RabbitMQ locally
  • Using tox