RabbitMQ with Python

Python developers connect to RabbitMQ using the pika library — the officially recommended AMQP client for Python. Pika is straightforward, well-documented, and supports all core RabbitMQ features including publisher confirms, manual acknowledgements, and TLS connections.

Installing Pika

pip install pika

Connecting to RabbitMQ

import pika

# Basic connection (localhost, default credentials)
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost')
)
channel = connection.channel()

For custom credentials and a specific vhost:

credentials = pika.PlainCredentials('myuser', 'mypassword')
parameters = pika.ConnectionParameters(
    host='rabbitmq.example.com',
    port=5672,
    virtual_host='/production',
    credentials=credentials,
    heartbeat=600,
    blocked_connection_timeout=300
)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

Publishing a Message

import pika, json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare a durable queue (idempotent — safe to call if it already exists)
channel.queue_declare(queue='orders', durable=True)

order = {'order_id': 1001, 'product': 'Widget', 'quantity': 5}

channel.basic_publish(
    exchange='',
    routing_key='orders',
    body=json.dumps(order),
    properties=pika.BasicProperties(
        content_type='application/json',
        delivery_mode=2  # persistent
    )
)
print("Order published")
connection.close()

Consuming Messages

import pika, json

def handle_order(ch, method, properties, body):
    order = json.loads(body.decode())
    print(f"Processing order {order['order_id']}: {order['quantity']}x {order['product']}")
    # Do actual processing here...
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='orders', durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='orders', on_message_callback=handle_order)

print("Waiting for orders. Press CTRL+C to stop.")
channel.start_consuming()

Error Handling and Retry

def handle_order(ch, method, properties, body):
    try:
        order = json.loads(body.decode())
        process_order(order)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except ValueError as e:
        # Bad message format — dead-letter it, do not requeue
        print(f"Invalid message: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    except Exception as e:
        # Transient error — requeue for retry
        print(f"Processing failed: {e}, requeueing")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

Asynchronous Consumer with SelectConnection

For production applications that need non-blocking I/O, use SelectConnection instead of BlockingConnection:

import pika

def on_message(channel, method, properties, body):
    print(f"Received: {body.decode()}")
    channel.basic_ack(delivery_tag=method.delivery_tag)

def on_channel_open(channel):
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue='orders', on_message_callback=on_message)

def on_connection_open(connection):
    connection.channel(on_open_callback=on_channel_open)

params = pika.ConnectionParameters('localhost')
connection = pika.SelectConnection(
    parameters=params,
    on_open_callback=on_connection_open
)

try:
    connection.ioloop.start()
except KeyboardInterrupt:
    connection.close()

Publisher Confirms in Python

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='critical-tasks', durable=True)
channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange='',
        routing_key='critical-tasks',
        body='Important task',
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print("Message confirmed by broker")
except pika.exceptions.UnroutableError:
    print("Message could not be routed!")

Working with Exchanges in Python

# Declare a topic exchange
channel.exchange_declare(exchange='app-events', exchange_type='topic', durable=True)

# Declare a queue and bind it
channel.queue_declare(queue='order-events', durable=True)
channel.queue_bind(exchange='app-events', queue='order-events', routing_key='order.#')

# Publish to the exchange
channel.basic_publish(
    exchange='app-events',
    routing_key='order.placed',
    body=json.dumps({'order_id': 2001}),
    properties=pika.BasicProperties(delivery_mode=2)
)

Connection Resilience

Production applications should reconnect automatically if the broker connection drops:

import pika, time

def connect():
    while True:
        try:
            connection = pika.BlockingConnection(
                pika.ConnectionParameters('localhost',
                    heartbeat=600,
                    blocked_connection_timeout=300)
            )
            return connection
        except pika.exceptions.AMQPConnectionError:
            print("Connection failed, retrying in 5 seconds...")
            time.sleep(5)

connection = connect()
channel = connection.channel()
# ... rest of setup

try:
    channel.start_consuming()
except pika.exceptions.ConnectionClosedByBroker:
    print("Connection closed by broker, reconnecting...")
    connection = connect()

Using aio-pika for Async Python

For asyncio-based applications, use aio-pika instead of pika:

pip install aio-pika

import asyncio, aio_pika

async def main():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("orders", durable=True)

        async with queue.iterator() as q:
            async for message in q:
                async with message.process():
                    print(f"Received: {message.body.decode()}")

asyncio.run(main())

Summary

Python connects to RabbitMQ using the pika library (synchronous) or aio-pika (asyncio). Use BlockingConnection for simple scripts and synchronous services. Use SelectConnection or aio-pika for production applications with non-blocking I/O. Always declare queues and exchanges as durable, publish messages as persistent, use manual acknowledgements, and set prefetch count for fair dispatch. Add reconnection logic to handle broker restarts and network interruptions gracefully.