rabbitmq-utils 1.6.0

Provide easy connection to RabbitMQ server.

rabbitmq-utils

A lightweight Python utility package to simplify working with RabbitMQ for producers, consumers, and RPC communication.

TLS Support

The following classes now support optional TLS (SSL) parameters for secure communication over port 5671:

• RabbitMQProducer, AsyncRabbitMQProducer
• RabbitMQConsumer, AsyncRabbitMQConsumer
• RPCClient
• RPCServer

You can enable TLS by passing the following optional parameters:

cafile = "path/to/ca.crt"
check_hostname = True

These should be provided only when the RabbitMQ server is configured with TLS (typically running on port 5671). If not provided, connections default to non-TLS on port 5672.

Producer

The RabbitMQProducer class allows you to send messages reliably to any RabbitMQ queue. It includes Publisher Confirms to ensure delivery success.

Note: To use the default exchange, pass an empty string ("") as the exchange name. When using the default exchange, the routing_key should be set to the queue name.

You can make the message persistent by setting persistent_message=True. This ensures that the message survives broker restarts at the cost of some performance.

For priority queues, use the priority=(some int) argument in send_message.

Example

from rabbitmq_utils import RabbitMQProducer
import json

message = json.dumps({'hello': 'world'})

rmqp = RabbitMQProducer(
    host='localhost', port=5672, virtual_host='/', 
    username='guest', password='guest', 
    exchange='test_exc', exchange_type='topic',
    persistent_message=False
)
is_sent = rmqp.send_message(
    message,
    routing_key='test_key'
)

if is_sent:
    print('INFO: Message sent.')
else:
    print('ERROR: Unable to send on desired routing key.')

Async Producer

The AsyncRabbitMQProducer class allows you to send messages asynchronously to RabbitMQ queues. It supports the same features as the synchronous producer but using async/await patterns.

Example

import asyncio
import json
from rabbitmq_utils import AsyncRabbitMQProducer

async def main():
    message = json.dumps({'hello': 'world'})
    
    producer = AsyncRabbitMQProducer(
        host='localhost', port=5672, virtual_host='/', 
        username='guest', password='guest', 
        exchange='test_exchange', exchange_type='topic',
        persistent_message=True
    )
    
    is_published, exc = await producer.publish_message(
        message,
        routing_key='test.routing.key',
        close_connection=True,
        return_exception=True
    )
    
    if is_published:
        print('Message published successfully')
    else:
        print(f'Failed to publish: {exc}')

asyncio.run(main())

With TLS

producer = AsyncRabbitMQProducer(
    host='localhost', port=5671, virtual_host='/', 
    username='guest', password='guest', 
    exchange='test_exchange', exchange_type='topic',
    cafile='path/to/ca.crt',
    check_hostname=True
)

Consumer

The RabbitMQConsumer class sets up a durable queue and exchange, binds them, and starts consuming messages using a user-defined callback function.

If you receive durability-related errors, try deleting the existing queue/exchange before re-running the code.

Sample Callback Function

def my_callback_function(ch, method, properties, body):
    message = body.decode()
    myLogic()  # your logic here
    ch.basic_ack(delivery_tag=method.delivery_tag)

Example

from rabbitmq_utils import RabbitMQConsumer

rmqc = RabbitMQConsumer(
    host='localhost', port=5672, virtual_host='/', 
    username='guest', password='guest', 
    queue_name='test_que', routing_key='test_key',
    exchange='test_exc', exchange_type='topic',
    callback_fun=my_callback_function,
    max_priority=2  # optional
)
rmqc.receive_message()

Async Consumer

The AsyncRabbitMQConsumer class provides asynchronous message consumption from RabbitMQ queues. It requires an async callback function to handle incoming messages.

Sample Async Callback Function

async def my_async_callback(body: str):
    message = json.loads(body)
    # your async logic here
    print(f'Received: {message}')

Example

import asyncio
import json
from rabbitmq_utils import AsyncRabbitMQConsumer

async def my_async_callback(body: str):
    message = json.loads(body)
    print(f'Received message: {message}')

async def main():
    consumer = AsyncRabbitMQConsumer(
        host='localhost', port=5672, virtual_host='/', 
        username='guest', password='guest', 
        queue_name='test_queue', routing_key='test.routing.key',
        exchange='test_exchange', exchange_type='topic',
        callback_fun=my_async_callback
    )
    
    await consumer.start_consuming()

asyncio.run(main())

With TLS

consumer = AsyncRabbitMQConsumer(
    host='localhost', port=5671, virtual_host='/', 
    username='guest', password='guest', 
    queue_name='test_queue', routing_key='test.routing.key',
    exchange='test_exchange', exchange_type='topic',
    callback_fun=my_async_callback,
    cafile='path/to/ca.crt',
    check_hostname=True
)

Remote Procedure Call (RPC)

RPC allow to run a function on a remote computer and wait for the result. It is a synchronous call. This package provide simplest implementation of RPC.

RPC consist of two parts. One is the server that will process the request and other is client that will generate the request to server. Following are example of RPC implementation.

Server

from rabbitmq_utils.rpc import RPCServer

# STARTING RPC SERVER
server = RPCServer(
        host='localhost', port=5672, virtual_host='/', 
        username='guest', password='guest', 
        queue_name='test_que', routing_key='test_key',
    	exchange='test_exc', exchange_type='topic',
    	callback_fun=rpc_callback_function
)
server.receive_message()

Callback function of RPC is different from consumer callback function. In this callback we will return the result back to client.

Note: result must be string. If it is not then use json.dumps(result) to convert it to string.

def rpc_callback_function(ch, method, properties, body):
    # GETTING MESSAGE
    message = body.decode()
    
    # PERFORM YOUR LOGIC HERE
    result = myLogic()
    
    # RETURING RESPONSE
    ch.basic_publish(
        exchange='', routing_key=properties.reply_to,
        properties=pika.BasicProperties(
            correlation_id = properties.correlation_id
        ),
        body=result
    )
    
    # ACKNOWLEDGE WORK IS DONE
    ch.basic_ack(delivery_tag=method.delivery_tag)
    return None

Client

from rabbitmq_utils.rpc import RPCClient
import json

# DEFINING MESSAGE
message = json.dumps({'hello': 'world'})

# SENDING
client = RPCClient(
    host='localhost', port=5672, virtual_host='/', 
    username='guest', password='guest', 
    exchange='test_exc', exchange_type='topic',
    timeout=3, # wait 3 seconds for response. default is None (infinite wait).
    persistent_message=False
)
is_sent, response = client.send_message(
    message,
    routing_key='test_key',
    return_response=True
)

# OUTPUT
print(f'is_sent: {is_sent} \t code: {client.getCode()} \t response: {response}')

Client send_message receive return_response argument (default=False). If this is True then client will wait for response for desire timeout period. You can receive response later if you want by using follow sample code:

# SEND REQUEST
is_sent = client.send_message(
    message,
    routing_key
)

# PERFORM YOUR LOGIC

# RECEIVE RESPONSE
response = client.receive_response()

Always check the validity of response using status code. Following code will help you check it:

status_code = client.getCode()
print(status_code)

Code is integer. Following table shows the meanings:

Code	Meaning
200	Response is successfully obtained.
408	Timeout occur.

Author

Tahir Rafique

Releases

Date	Version	Summary
07-Jun-26	1.6.0	Adding async support for producer and consumer with TLS.
29-Jun-25	1.5.0	Adding TLS support for rabbitmq running on 5671 port.
17-Jan-24	1.4.1	Adding exception handling in default callback function of consumer.
13-Dec-23	1.4.0	Adding queue priority in producer and consumer.
14-Jul-23	1.3.0	Adding persistent message option.
14-Jul-23	1.2.1	Correcting documentation.
21-Jun-23	1.2.0	Adding RPC to module.
27-Apr-23	1.0.1	Improving default callback function.
27-Apr-23	1.0.0	Initial build