RabbitMQ: Utiliser Python pour se connecter à RabbitMQ
Cette documentation fait partie du guide Premiers pas. Consultez le guide complet ici : Comment utiliser Stackhero pour RabbitMQ.
Voyons ensemble comment connecter votre application Python à RabbitMQ en utilisant la bibliothèque Aio Pika. Dans la plupart des cas, il suffit de fournir l'URL AMQPS pour établir une connexion sécurisée :
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>",
)
Voici un exemple complet qui montre comment établir une connexion sécurisée à RabbitMQ, créer un channel et déclarer une file de base. C'est une excellente façon de vérifier votre configuration :
import asyncio
import logging
import aio_pika
async def main() -> None:
# Si vous souhaitez afficher les logs de debug, vous pouvez décommenter la ligne suivante
# 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())
Gérer l'erreur unable to get local issuer certificate
Si vous rencontrez une erreur de ce type lors de la connexion depuis Python :
aiormq.exceptions.AMQPConnectionError: [Errno 5] [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1006)
cela signifie généralement que le certificat CA Let's Encrypt manque sur votre système. Pour corriger cela, installez le paquet de certificats CA adapté à votre système d'exploitation :
-
Sur Ubuntu/Debian, exécutez :
sudo apt install ca-certificates -
Sur Alpine Linux, exécutez :
apk add ca-certificates
Si vous ne pouvez pas utiliser ces commandes, vous pouvez aussi installer le certificat CA manuellement :
-
Téléchargez le certificat CA Let's Encrypt depuis https://letsencrypt.org/certs/isrgrootx1.pem.
-
Ensuite, connectez-vous à RabbitMQ dans votre code Python en spécifiant le fichier du certificat CA :
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 )
Voici un exemple complet utilisant le certificat CA Let's Encrypt pour des connexions sécurisées :
import asyncio
import logging
import ssl
import aio_pika
async def main() -> None:
# Pour activer les logs de debug, vous pouvez décommenter cette ligne
# logging.basicConfig(level=logging.DEBUG)
ssl_context = ssl.create_default_context()
# Chargez le certificat CA Let's Encrypt que vous avez téléchargé
# Par exemple : 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())