RabbitMQ: Using Python to connect to RabbitMQ
This documentation is part of the Getting started guide. View the full guide here: How to use Stackhero for RabbitMQ.
Let us walk through how you can connect your Python application to RabbitMQ using the Aio Pika library. In most cases, you only need to provide the AMQPS URL for a secure connection:
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>",
)
Here is a complete example that shows how you can establish a secure connection to RabbitMQ, create a channel, and declare a basic queue. This is a great way to verify your setup:
import asyncio
import logging
import aio_pika
async def main() -> None:
# If you want to see debug logs, you can uncomment the next line
# logging.basicConfig(level=logging.DEBUG)
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>"
)
async with connection:
print("The connection worked!")
channel = await connection.channel()
await channel.set_qos(prefetch_count=10)
queue = await channel.declare_queue("test_queue", auto_delete=True)
if __name__ == "__main__":
asyncio.run(main())
Handling error unable to get local issuer certificate
If you encounter an error like this when connecting from Python:
aiormq.exceptions.AMQPConnectionError: [Errno 5] [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1006)
it typically means your system is missing the Let's Encrypt CA certificate. To fix this, you can install the common CA certificates package for your operating system:
-
On Ubuntu/Debian, you can run:
sudo apt install ca-certificates -
On Alpine Linux, you can run:
apk add ca-certificates
If you are unable to use these commands, you still have the option to install the CA certificate manually:
-
Download the Let's Encrypt CA certificate from https://letsencrypt.org/certs/isrgrootx1.pem.
-
Then, you can connect to RabbitMQ in your Python code by specifying the CA certificate file:
import ssl ssl_context = ssl.create_default_context() ssl_context.load_verify_locations(cafile='isrgrootx1.pem') connection = await aio_pika.connect_robust( "amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>", ssl_context=ssl_context )
Here is a full example using the Let's Encrypt CA certificate for secure connections:
import asyncio
import logging
import ssl
import aio_pika
async def main() -> None:
# To enable debug logs, you can uncomment this line
# logging.basicConfig(level=logging.DEBUG)
ssl_context = ssl.create_default_context()
# Load the Let's Encrypt CA certificate you downloaded
# For example: wget https://letsencrypt.org/certs/isrgrootx1.pem
ssl_context.load_verify_locations(cafile='isrgrootx1.pem')
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>",
ssl_context=ssl_context
)
async with connection:
print("The connection worked!")
channel = await connection.channel()
await channel.set_qos(prefetch_count=10)
queue = await channel.declare_queue("test_queue", auto_delete=True)
if __name__ == "__main__":
asyncio.run(main())