Python with RabbitMQ and LavinMQ

We recommend Pika, the most widely used Python client for AMQP 0-9-1.

pip install pika

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the examples connect to a local broker.

Save the connection helper as amqp.py:

import os

import pika


def connect():
    url = os.environ.get("BROKER_URL", "amqp://guest:guest@localhost/%2f")
    return pika.BlockingConnection(pika.URLParameters(url))

Save the publisher as publisher.py:

import pika

import amqp


def publish(channel, queue, message):
    channel.queue_declare(queue=queue, durable=True)
    channel.confirm_delivery()
    channel.basic_publish(
        exchange="",
        routing_key=queue,
        body=message,
        properties=pika.BasicProperties(delivery_mode=pika.DeliveryMode.Persistent),
    )
    print(f" [x] Sent {message!r}")


if __name__ == "__main__":
    with amqp.connect() as connection:
        publish(connection.channel(), "hello", "Hello world!")

Save the consumer as worker.py in the same directory:

import amqp


def consume(channel, queue, handler):
    channel.queue_declare(queue=queue, durable=True)
    channel.basic_qos(prefetch_count=10)

    def on_message(ch, method, properties, body):
        handler(body.decode())
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(queue=queue, on_message_callback=on_message)


if __name__ == "__main__":
    with amqp.connect() as connection:
        channel = connection.channel()
        consume(channel, "hello", lambda message: print(f" [x] Received {message!r}"))
        print(" [*] Waiting for messages. To exit press CTRL+C")
        try:
            channel.start_consuming()
        except KeyboardInterrupt:
            pass

Run python worker.py in one terminal and python publisher.py in another. Keep handlers short: BlockingConnection cannot service heartbeats while a handler is running. For long tasks, use a separate thread and schedule the ack with connection.add_callback_threadsafe, as shown in Pika’s threaded consumer example.

A runnable example is in python/pika in cloudamqp/examples.

Good to know

Connect with the URL. pika.BlockingConnection(pika.URLParameters(url)), where the URL is the one on your instance’s details page in the CloudAMQP console. With amqps:// the server certificate and hostname are verified.
Use publisher confirms (confirm_delivery()) and persistent messages, so a message isn’t lost if the connection or broker goes down.
Ack after processing, and set a prefetch limit with basic_qos, so a consumer doesn’t buffer more messages than it can handle.
Pika doesn’t reconnect. Catch pika.exceptions.AMQPConnectionError and connect again, or let your process manager restart the process.
One connection per thread. A BlockingConnection isn’t thread-safe.

Other connection parameters are listed in Pika’s documentation.

Other Python libraries

aio-pika, an asyncio client that reconnects automatically
rabbitpy, a thread-safe client
Celery, a task queue that can use CloudAMQP as its broker

← Ruby

Celery →