AMQPStorm

2.11.1 · active · verified Sun Apr 12

AMQPStorm is a thread-safe Python client library for RabbitMQ, offering comprehensive features for both messaging and RabbitMQ management. Currently at version 2.11.1, the library maintains an active release cadence with regular updates and improvements, ensuring stability and compatibility.

Warnings

breaking Starting with version 2.0.0, messages are delivered as `Message` objects by default. Previous versions might have returned tuples or dictionaries directly from consumer functions.
Fix: Update consumer logic to expect and interact with `amqpstorm.Message` objects. Access message content via `message.body` and properties via `message.properties` or `message.get_property('key')`. If necessary, explicitly set `to_tuple=True` or `to_dict=True` when consuming, but using `Message` objects is recommended.
gotcha There's an open issue (#144) regarding `Import Error due to API Change in pamqp>=3.0.0 (specification -> commands)`, which is a core dependency. This might cause issues with newer `pamqp` versions.
Fix: If encountering `Import Error` related to `pamqp`, consider pinning your `pamqp` version to `<3.0.0` or checking the AMQPStorm GitHub issues for a resolution or newer AMQPStorm version that addresses this compatibility. `pip install 'pamqp<3.0.0'`
deprecated The library's `amqpstorm/message.py` currently uses `datetime.utcnow()`, which is deprecated in Python 3.12+ in favor of `datetime.now(timezone.utc)`. This is an open issue (#143).
Fix: While this is an internal library usage, be aware that warnings might appear in Python 3.12+ environments. Keep an eye on AMQPStorm updates for a fix. No direct user-level fix is available, but it's generally harmless beyond the warning.
gotcha Properly closing connections and channels is crucial to avoid resource leaks or hangs, especially in long-running applications or when handling errors.
Fix: Always use `with` statements (context managers) for `Connection` and `Channel` objects, as shown in the quickstart. This ensures resources are automatically released, even if errors occur. Explicitly call `connection.close()` and `channel.close()` if context managers are not used.
gotcha When using SSL/TLS connections, incorrect `ssl_options` or `verify` parameter settings can lead to insecure connections (e.g., no certificate verification) or failed connections (e.g., incorrect CA bundles).
Fix: Carefully configure `ssl_options` and `verify`. For production, always use `verify=True` (or provide a path to a CA bundle) and ensure `server_hostname` is correctly set. Only use `verify=False` for testing purposes where the risk is understood and mitigated.

Install

pip install amqpstorm Core library
pip install amqpstorm[management] With Management API features
pip install amqpstorm[pool] With connection pooling features

Imports

Connection
```
from amqpstorm import Connection
```
UriConnection
```
from amqpstorm import UriConnection
```
Message
```
from amqpstorm import Message
```
management
```
from amqpstorm import management
```
Provides access to the Management API classes like ManagementApi.

Quickstart

This quickstart demonstrates how to establish a connection to RabbitMQ, open a channel, declare a queue, and publish a basic message using AMQPStorm. It uses environment variables for connection details and context managers for robust resource management.

import amqpstorm
import os

RABBITMQ_HOST = os.environ.get('RABBITMQ_HOST', 'localhost')
RABBITMQ_USER = os.environ.get('RABBITMQ_USER', 'guest')
RABBITMQ_PASS = os.environ.get('RABBITMQ_PASS', 'guest')

try:
    # Establish a connection using a context manager for proper resource handling
    with amqpstorm.Connection(RABBITMQ_HOST, RABBITMQ_USER, RABBITMQ_PASS) as connection:
        # Open a channel using a context manager
        with connection.channel() as channel:
            # Declare a queue (idempotent operation)
            channel.queue.declare('my_queue')

            # Publish a simple message to 'my_queue'
            channel.basic.publish(body='Hello, RabbitMQ!', routing_key='my_queue')
            print(" [x] Sent 'Hello, RabbitMQ!'")
except amqpstorm.AMQPConnectionError as e:
    print(f"Error connecting to RabbitMQ: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")

view raw JSON →