utils.py

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)