Python with RabbitMQ—simplified so you won't have to.
Project description
Features
Stop worrying about boilerplating and implementing retry logic for your queues. PyRMQ already does it for you.
- Use out-of-the-box
Consumer
andPublisher
classes created frompika
for your projects and tests. - Custom DLX-DLK-based retry logic for message consumption.
- Message priorities
- Works with Python 3.
- Production ready
Getting Started
Installation
PyRMQ is available at PyPi.
pip install pyrmq
Usage
Publishing
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: 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",
callback=callback
)
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
.
Using other exchange types
You can use another exchange type just by simply specifying it in the Publisher class. The default is
direct
.
from pyrmq import Publisher
queue_args = {"routing.sample": "sample", "x-match": "all"}
publisher = Publisher(
exchange_name="exchange_name",
exchange_type="headers",
queue_args=queue_args
)
message_properties = {"headers": {"routing.sample": "sample"}}
publisher.publish({"pyrmq": "My first message"}, message_properties=message_properties)
This is an example of how to publish to a headers exchange that will get routed based on its headers.
Documentation
Visit https://pyrmq.readthedocs.io for the most up-to-date documentation.
Testing
For development, just run:
pytest
To test for all the supported Python versions:
pip install tox
tox
To test for a specific Python version:
tox -e py38
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.