rabbitpy

To perform concurrent operations like publishing and consuming in separate threads, pass the rabbitpy.Connection object to your worker threads. Inside each thread, use connection.channel() to create a thread-local channel. This ensures that each thread operates on its own isolated AMQP channel, maintaining protocol safety.

import rabbitpy
import threading

EXCHANGE = 'threading_example'
QUEUE = 'threading_queue'
ROUTING_KEY = 'test'
MESSAGE_COUNT = 100


def consumer(connection):
    """Consume MESSAGE_COUNT messages then exit."""
    received = 0
    # Create a dedicated channel for this thread
    with connection.channel() as channel:
        for message in rabbitpy.Queue(channel, QUEUE):
            print(message.body)
            message.ack()
            received += 1
            if received == MESSAGE_COUNT:
                break


def publisher(connection):
    """Publish MESSAGE_COUNT messages."""
    # Create a dedicated channel for this thread
    with connection.channel() as channel:
        for index in range(MESSAGE_COUNT):
            message = rabbitpy.Message(channel, f'Message #{index}')
            message.publish(EXCHANGE, ROUTING_KEY)


with rabbitpy.Connection() as connection:
    # Setup topology in the main thread
    with connection.channel() as channel:
        exchange = rabbitpy.Exchange(channel, EXCHANGE)
        exchange.declare()

        queue = rabbitpy.Queue(channel, QUEUE)
        queue.declare()
        queue.bind(EXCHANGE, ROUTING_KEY)

    kwargs = {'connection': connection}

    consumer_thread = threading.Thread(target=consumer, kwargs=kwargs)
    consumer_thread.start()

    publisher_thread = threading.Thread(target=publisher, kwargs=kwargs)
    publisher_thread.start()

    consumer_thread.join()

rabbitpy uses two categories of exceptions:

1. rabbitpy Exceptions: Specific to the library's logic and state.
2. AMQP Exceptions: Raised when RabbitMQ sends a channel or connection close with a specific AMQP reply code (e.g., when a precondition fails during queue declaration).

You should wrap your connection and channel operations in try/except blocks to handle these errors, especially when performing operations like declaring queues with specific parameters that might conflict with existing ones.

import rabbitpy

try:
    with rabbitpy.Connection() as connection:
        with connection.channel() as channel:
            queue = rabbitpy.Queue(channel, 'exception-test')
            queue.durable = True
            queue.declare()
            # Changing parameters and re-declaring will trigger an exception
            queue.durable = False
            queue.declare()  # raises rabbitpy.exceptions.AMQPPreconditionFailed
except rabbitpy.exceptions.AMQPPreconditionFailed:
    print('Queue already exists with different parameters')

rabbitpy.Connection objects provide a thread-safe connection to RabbitMQ used for authentication and executing channel-based RPC commands. You can specify connection details using the AMQP URI syntax.

Connections can be managed in two ways:

1. Manual management: Create the object and call .close() manually.
2. Context manager: Use the with statement to ensure the connection is automatically closed when the block exits.

Note on Heartbeats: If heartbeats are enabled (default is 300 seconds), a ConnectionResetException will be raised if RabbitMQ does not send a heartbeat within $\ge 2$ heartbeat intervals.

# Manual management
conn = rabbitpy.Connection('amqp://guest:guest@localhost:5672/%2F')
conn.close()

# Context manager (Recommended)
with rabbitpy.Connection('amqp://guest:guest@localhost:5672/%2F') as conn:
    with conn.channel() as channel:
        # use channel
        pass

The Message class is the primary interface for interacting with RabbitMQ messages. It is used both to create messages for publishing and to represent messages received from the broker.

Key Capabilities

• Publishing: Send messages to an exchange with a routing key.
• Consuming: Access metadata like delivery_tag, routing_key, and exchange from received messages.
• Acknowledgment: Use ack(), nack(), or reject() to manage message lifecycle after consumption.
• Serialization: Automatically serializes dict or list bodies to JSON and sets content_type to application/json.
• Opinionated Mode: When opinionated=True, the class automatically generates a message_id (UUID) and a timestamp if they are not provided.

Message Properties

You can pass a dictionary of properties to the constructor. Note that the AMQP type property is referred to as message_type in rabbitpy to avoid Python keyword conflicts.

Available properties:

• app_id
• content_type
• content_encoding
• correlation_id
• delivery_mode
• expiration
• headers
• message_id
• message_type
• priority
• reply_to
• timestamp
• user_id

# Example: Creating and publishing a message
from rabbitpy import Message

msg = Message(
    channel=channel,
    body_value={'key': 'value'},
    properties={'app_id': 'my_app'},
    opinionated=True
)
msg.publish(exchange=my_exchange, routing_key='my_key')

The core API for rabbitpy follows a hierarchical structure: Connection -> Channel -> Exchange/Queue.

• Connection: Manages the physical connection to the RabbitMQ broker.
• Channel: A virtual connection within a Connection where most AMQP operations occur.
• Exchange: An entity that receives messages and routes them to queues. Supported types include DirectExchange, FanoutExchange, HeadersExchange, and TopicExchange.
• Queue: A buffer that stores messages.
• Message: The data unit being sent or received.
• AMQP: The underlying AMQP implementation layer.

from rabbitpy import Connection, Channel, DirectExchange, Queue, Message

# Typical workflow pattern:
connection = Connection(host='localhost')
with connection:
    channel = connection.channel()
    exchange = channel.exchange('my_exchange', type='direct')
    queue = channel.queue('my_queue')
    queue.bind(exchange)
    
    message = Message(body='Hello World')
    exchange.publish(message)
    
    # Consuming
    for msg in queue.consume():
        print(msg.body)