# Python with RabbitMQ and LavinMQ

We recommend [Pika](https://pika.readthedocs.io), the most widely used Python client for AMQP 0-9-1.

```bash
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`:

```python
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`:

```python
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:

```python
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](https://github.com/pika/pika/blob/main/examples/basic_consumer_threaded.py).

A runnable example is in [python/pika in cloudamqp/examples](https://github.com/cloudamqp/examples/tree/main/python/pika).

## 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](https://pika.readthedocs.io/en/stable/modules/parameters.html).

## Other Python libraries

- [aio-pika](https://github.com/mosquito/aio-pika), an asyncio client that reconnects automatically
- [rabbitpy](https://github.com/gmr/rabbitpy), a thread-safe client
- [Celery](/docs/celery/), a task queue that can use CloudAMQP as its broker