import logging import os import pika logger = logging.getLogger(__name__) logging.getLogger("pika").setLevel(logging.WARNING) BROKER_ADDRESS = os.getenv('BROKER_ADDRESS') or 'localhost' BROKER_USER = os.getenv('BROKER_USER') or 'guest' BROKER_PASSWORD = os.getenv('BROKER_PASSWORD') or 'guest' BROKER_PORT = os.getenv('BROKER_PORT') or '5672' BROKER_URL = f'amqp://{BROKER_USER}:{BROKER_PASSWORD}@{BROKER_ADDRESS}:{BROKER_PORT}/%2F' def test_connection(): try: # Create a connection to the broker connection = get_blocking_connection() # Declare a channel and a queue channel = connection.channel() # Close the connection connection.close() except Exception as e: logger.error(f"Failed to connect to broker; message: {e}") raise def get_blocking_connection(): credentials = pika.PlainCredentials(username=BROKER_USER, password=BROKER_PASSWORD) parameters = pika.ConnectionParameters(host=BROKER_ADDRESS, credentials=credentials) return pika.BlockingConnection(parameters)