amqp/service/create_exchange.py

@@ -0,0 +1,90 @@
+import asyncio
+import aio_pika
+import os
+import argparse
+import sys
+
+from aio_pika.abc import AbstractRobustExchange
+
+
+async def create_exchange(loop, rabbit_user, rabbit_password, rabbit_host):
+    """
+    Asynchronously connects to RabbitMQ, declares a direct exchange,
+    and prints the result.  Handles potential connection errors.
+    """
+    exchange_name = "cleverthis.clevermicro.management"
+    connection = None  # Keep track of the connection
+
+    try:
+        # Construct the connection URI from environment variables
+        connection_uri = f"amqp://{rabbit_user}:{rabbit_password}@{rabbit_host}/"
+
+        # Attempt to establish the connection
+        connection = await aio_pika.connect_robust(connection_uri, loop=loop)
+        print(f"Connected to RabbitMQ: {connection_uri}")
+
+        # Create a channel
+        channel = await connection.channel()
+
+        # Declare the exchange
+        exchange: AbstractRobustExchange = await channel.declare_exchange(
+            exchange_name, aio_pika.ExchangeType.DIRECT
+        )
+        print(f"Exchange '{exchange.name}' created successfully.")
+
+    except aio_pika.exceptions.AMQPConnectionError as e:
+        print(f"Error: Could not connect to RabbitMQ.  Details: {e}")
+        sys.exit(1)  # Exit with an error code
+    except aio_pika.exceptions.AMQPChannelError as e:
+        print(f"Error creating channel or exchange: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+        sys.exit(1)
+    finally:
+        # Ensure the connection is closed, even if errors occur
+        if connection:
+            await connection.close()
+            print("Connection to RabbitMQ closed.")
+
+def main():
+    """
+    Main function to parse command-line arguments, retrieve RabbitMQ
+    credentials from environment variables, and run the
+    async exchange creation.
+    """
+    # Set up argument parsing
+    parser = argparse.ArgumentParser(description="Create a RabbitMQ exchange.")
+    parser.add_argument(
+        "--host",
+        default="localhost",
+        help="RabbitMQ host (default: localhost)",
+    )
+
+    # Parse the arguments
+    args = parser.parse_args()
+    rabbit_host = args.host
+
+    # Get RabbitMQ credentials from environment variables
+    rabbit_user = os.environ.get("RABBIT_MQ_USER")
+    rabbit_password = os.environ.get("RABBIT_MQ_PASSWORD")
+
+    if not rabbit_user or not rabbit_password:
+        print(
+            "Error: RabbitMQ credentials not found in environment variables.\n"
+            "Please set RABBIT_MQ_USER and RABBIT_MQ_PASSWORD."
+        )
+        sys.exit(1)  # Exit with an error code
+
+    # Create the event loop
+    loop = asyncio.get_event_loop()
+
+    # Run the asynchronous exchange creation
+    loop.run_until_complete(create_exchange(loop, rabbit_user, rabbit_password, rabbit_host))
+
+    # Close the loop
+    loop.close()
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.23-dev"
-description = "The Python implementation of the AMQ Adaptor for CleverMicro"
+version = "0.2.24"
+description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -21,13 +21,13 @@ requires-python = ">=3.11"
 dependencies = [
     "aio-pika~=9.5.5",
     "aiohttp~=3.11.11",
-    "aio_pika~=9.5.5",
     "fastapi==0.114.2",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
+    "pika~=1.3.2",
     "python_multipart"
 ]