aiormq: Asynchronous AMQP Client

Warnings

• breaking Python 3.7 support was dropped in aiormq version 6.8.0. If you are using Python 3.7, you must upgrade your Python version or pin `aiormq<6.8.0`.
  Fix: Upgrade to Python 3.8+ or pin `aiormq` to a version less than 6.8.0 in your `requirements.txt`.
• gotcha Properly closing connections and channels is crucial in asynchronous applications to prevent resource leaks and ensure graceful shutdown. Failing to `await connection.close()` and `await channel.close()` can lead to hung connections or unreleased resources.
  Fix: Always wrap your AMQP operations in `try...finally` blocks to ensure `await connection.close()` and `await channel.close()` are called, even if exceptions occur. For channels, ensure they are closed before the connection.
• gotcha While `aiormq` aims to handle AMQP URLs robustly, be mindful of special characters, complex VHosts, or non-standard query parameters in your connection string. Historically, there have been minor fixes related to URL parsing (e.g., slash unquoting in 6.8.1).
  Fix: Ensure your AMQP connection URL is properly URL-encoded, especially for parts like VHosts or usernames/passwords that might contain special characters. Test complex URLs thoroughly.

Install

• pip install aiormq Install aiormq

Imports

• connect

  from aiormq import connect

• Channel

  from aiormq.abc import Channel

  While Channel is part of the internal structure, for type hinting, import from `aiormq.abc`.
• IncomingMessage

  from aiormq.abc import IncomingMessage

  Required for type hinting message callbacks during consumption.

Quickstart

This quickstart demonstrates how to connect to an AMQP broker (like RabbitMQ), declare a queue, publish a message, and consume messages asynchronously. It uses `os.environ.get` for the AMQP URL, making it easy to configure without hardcoding credentials. Remember to have a RabbitMQ instance running at `localhost:5672` or specify a different `AMQP_URL`.

import asyncio
import os
from aiormq import connect
from aiormq.abc import IncomingMessage

# Get AMQP URL from environment variable, default to local RabbitMQ
AMQP_URL = os.environ.get('AMQP_URL', 'amqp://guest:guest@localhost/')
QUEUE_NAME = 'aiormq_test_queue'

async def on_message(message: IncomingMessage):
    """Callback for consuming messages."""
    print(f"[x] Received: {message.body.decode()}")
    await message.ack() # Acknowledge the message

async def main():
    connection = None
    try:
        # Establish connection
        print(f"[*] Connecting to {AMQP_URL}...")
        connection = await connect(AMQP_URL)
        print("[*] Connection established.")

        # Create a channel
        channel = await connection.channel()
        print("[*] Channel created.")

        # Declare a queue (idempotent operation)
        await channel.queue_declare(QUEUE_NAME)
        print(f"[*] Queue '{QUEUE_NAME}' declared.")

        # Publish a message
        message_body = b"Hello, aiormq world!"
        await channel.basic_publish(
            exchange='',
            routing_key=QUEUE_NAME,
            body=message_body
        )
        print(f"[x] Published message: '{message_body.decode()}'")

        # Start consuming messages
        consumer_tag = await channel.basic_consume(QUEUE_NAME, on_message)
        print(f"[*] Consuming from '{QUEUE_NAME}'. Consumer tag: {consumer_tag}")

        # Keep the consumer running for a short period (e.g., 5 seconds)
        print("[*] Waiting for messages... Press Ctrl+C to exit")
        await asyncio.sleep(5)

        # Stop consuming
        await channel.basic_cancel(consumer_tag)
        print(f"[*] Consumer '{consumer_tag}' cancelled.")

    except Exception as e:
        print(f"[!] An error occurred: {e}")
    finally:
        if connection:
            print("[*] Closing connection...")
            await connection.close()
            print("[*] Connection closed.")

if __name__ == '__main__':
    asyncio.run(main())