API

Basics

There are two principal objects when using aioamqp:

  • The protocol object, used to begin a connection to aioamqp,
  • The channel object, used when creating a new channel to effectively use an AMQP channel.

Starting a connection

Starting a connection to AMQP really mean instanciate a new asyncio Protocol subclass.

aioamqp.connect(host, port, login, password, virtualhost, ssl, login_method, insist, protocol_factory, verify_ssl, loop, kwargs) → Transport, AmqpProtocol

Convenient method to connect to an AMQP broker

Parameters:
  • host (str) – the host to connect to
  • port (int) – broker port
  • login (str) – login
  • password (str) – password
  • virtualhost (str) – AMQP virtualhost to use for this connection
  • ssl (bool) – create an SSL connection instead of a plain unencrypted one
  • verify_ssl (bool) – verify server’s SSL certificate (True by default)
  • login_method (str) – AMQP auth method
  • insist (bool) – insist on connecting to a server
  • protocol_factory (AmqpProtocol) – factory to use, if you need to subclass AmqpProtocol
  • loop (EventLopp) – set the event loop to use
  • kwargs (dict) – arguments to be given to the protocol_factory instance
import asyncio
import aioamqp

async def connect():
    try:
        transport, protocol = await aioamqp.connect()  # use default parameters
    except aioamqp.AmqpClosedConnection:
        print("closed connections")
        return

    print("connected !")
    await asyncio.sleep(1)

    print("close connection")
    await protocol.close()
    transport.close()

asyncio.get_event_loop().run_until_complete(connect())

In this example, we just use the method “start_connection” to begin a communication with the server, which deals with credentials and connection tunning.

If you’re not using the default event loop (e.g. because you’re using aioamqp from a different thread), call aioamqp.connect(loop=your_loop).

The AmqpProtocol uses the kwargs arguments to configure the connection to the AMQP Broker:

AmqpProtocol.__init__(self, *args, **kwargs):

The protocol to communicate with AMQP

Parameters:
  • channel_max (int) – specifies highest channel number that the server permits. Usable channel numbers are in the range 1..channel-max. Zero indicates no specified limit.
  • frame_max (int) – the largest frame size that the server proposes for the connection, including frame header and end-byte. The client can negotiate a lower value. Zero means that the server does not impose any specific limit but may reject very large frames if it cannot allocate resources for them.
  • heartbeat (int) – the delay, in seconds, of the connection heartbeat that the server wants. Zero means the server does not want a heartbeat.
  • loop (Asyncio.EventLoop) – specify the eventloop to use.
  • client_properties (dict) – configure the client to connect to the AMQP server.

Handling errors

The connect() method has an extra ‘on_error’ kwarg option. This on_error is a callback or a coroutine function which is called with an exception as the argument:

import asyncio
import socket
import aioamqp

async def error_callback(exception):
    print(exception)

async def connect():
    try:
        transport, protocol = await aioamqp.connect(
            host='nonexistant.com',
            on_error=error_callback,
            client_properties={
                'program_name': "test",
                'hostname' : socket.gethostname(),
            },

        )
    except aioamqp.AmqpClosedConnection:
        print("closed connections")
        return

asyncio.get_event_loop().run_until_complete(connect())

Publishing messages

A channel is the main object when you want to send message to an exchange, or to consume message from a queue:

channel = await protocol.channel()

When you want to produce some content, you declare a queue then publish message into it:

await channel.queue_declare("my_queue")
await channel.publish("aioamqp hello", '', "my_queue")

Note: we’re pushing message to “my_queue” queue, through the default amqp exchange.

Consuming messages

When consuming message, you connect to the same queue you previously created:

import asyncio
import aioamqp

async def callback(channel, body, envelope, properties):
    print(body)

channel = await protocol.channel()
await channel.basic_consume(callback, queue_name="my_queue")

The basic_consume method tells the server to send us the messages, and will call callback with amqp response arguments.

The consumer_tag is the id of your consumer, and the delivery_tag is the tag used if you want to acknowledge the message.

In the callback:

  • the first body parameter is the message

  • the envelope is an instance of envelope.Envelope class which encapsulate a group of amqp parameter such as:

    consumer_tag
    delivery_tag
    exchange_name
    routing_key
    is_redeliver
    
  • the properties are message properties, an instance of properties.Properties with the following members:

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

Server Cancellation

RabbitMQ offers an AMQP extension to notify a consumer when a queue is deleted. See Consumer Cancel Notification for additional details. aioamqp enables the extension for all channels but takes no action when the consumer is cancelled. Your application can be notified of consumer cancellations by adding a callback to the channel:

async def consumer_cancelled(channel, consumer_tag):
    # implement required cleanup here
    pass


async def consumer(channel, body, envelope, properties):
    await channel.basic_client_ack(envelope.delivery_tag)


channel = await protocol.channel()
channel.add_cancellation_callback(consumer_cancelled)
await channel.basic_consume(consumer, queue_name="my_queue")

The callback can be a simple callable or an asynchronous co-routine. It can be used to restart consumption on the channel, close the channel, or anything else that is appropriate for your application.

Queues

Queues are managed from the Channel object.

Channel.queue_declare(queue_name, passive, durable, exclusive, auto_delete, no_wait, arguments, timeout) → dict

Coroutine, creates or checks a queue on the broker

Parameters:
  • queue_name (str) – the queue to receive message from
  • passive (bool) – if set, the server will reply with Declare-Ok if the queue already exists with the same name, and raise an error if not. Checks for the same parameter as well.
  • durable (bool) – if set when creating a new queue, the queue will be marked as durable. Durable queues remain active when a server restarts.
  • exclusive (bool) – request exclusive consumer access, meaning only this consumer can access the queue
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the queue.
  • timeout (int) – wait for the server to respond after timeout

Here is an example to create a randomly named queue with special arguments x-max-priority:

result = await channel.queue_declare(
    queue_name='', durable=True, arguments={'x-max-priority': 4}
)
Channel.queue_delete(queue_name, if_unused, if_empty, no_wait, timeout)

Coroutine, delete a queue on the broker

Parameters:
  • queue_name (str) – the queue to receive message from
  • if_unused (bool) – the queue is deleted if it has no consumers. Raise if not.
  • if_empty (bool) – the queue is deleted if it has no messages. Raise if not.
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the queue.
  • timeout (int) – wait for the server to respond after timeout
Channel.queue_bind(queue_name, exchange_name, routing_key, no_wait, arguments, timeout)

Coroutine, bind a queue to an exchange

Parameters:
  • queue_name (str) – the queue to receive message from.
  • exchange_name (str) – the exchange to bind the queue to.
  • routing_key (str) – the routing_key to route message.
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the queue.
  • timeout (int) – wait for the server to respond after timeout

This simple example creates a queue, an exchange and bind them together.

channel = await protocol.channel()
await channel.queue_declare(queue_name='queue')
await channel.exchange_declare(exchange_name='exchange')

await channel.queue_bind('queue', 'exchange', routing_key='')
Channel.queue_unbind(queue_name, exchange_name, routing_key, arguments, timeout)

Coroutine, unbind a queue and an exchange.

Parameters:
  • queue_name (str) – the queue to receive message from.
  • exchange_name (str) – the exchange to bind the queue to.
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the queue.
  • timeout (int) – wait for the server to respond after timeout
PARAM STR ROUTING_KEY:
 

THE ROUTING_KEY TO ROUTE MESSAGE.

Channel.queue_purge(queue_name, no_wait, timeout)

Coroutine, purge a queue

Parameters:queue_name (str) – the queue to receive message from.

Exchanges

Exchanges are used to correctly route message to queue: a publisher publishes a message into an exchanges, which routes the message to the corresponding queue.

Channel.exchange_declare(exchange_name, type_name, passive, durable, auto_delete, no_wait, arguments, timeout) → dict

Coroutine, creates or checks an exchange on the broker

Parameters:
  • exchange_name (str) – the exchange to receive message from
  • type_name (str) – the exchange type (fanout, direct, topics …)
  • passive (bool) – if set, the server will reply with Declare-Ok if the exchange already exists with the same name, and raise an error if not. Checks for the same parameter as well.
  • durable (bool) – if set when creating a new exchange, the exchange will be marked as durable. Durable exchanges remain active when a server restarts.
  • auto_delete (bool) – if set, the exchange is deleted when all queues have finished using it.
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the exchange.
  • timeout (int) – wait for the server to respond after timeout

Note: the internal flag is deprecated and not used in this library.

channel = await protocol.channel()
await channel.exchange_declare(exchange_name='exchange', auto_delete=True)
Channel.exchange_delete(exchange_name, if_unused, no_wait, timeout)

Coroutine, delete a exchange on the broker

Parameters:
  • exchange_name (str) – the exchange to receive message from
  • if_unused (bool) – the exchange is deleted if it has no consumers. Raise if not.
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the exchange.
  • timeout (int) – wait for the server to respond after timeout
Channel.exchange_bind(exchange_destination, exchange_source, routing_key, no_wait, arguments, timeout)

Coroutine, binds two exchanges together

Parameters:
  • exchange_destination (str) – specifies the name of the destination exchange to bind
  • exchange_source (str) – specified the name of the source exchange to bind.
  • exchange_destination – specifies the name of the destination exchange to bind
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the exchange.
  • timeout (int) – wait for the server to respond after timeout
Channel.exchange_unbind(exchange_destination, exchange_source, routing_key, no_wait, arguments, timeout)
Coroutine, unbind an exchange from an exchange.
Parameters:
  • exchange_destination (str) – specifies the name of the destination exchange to bind
  • exchange_source (str) – specified the name of the source exchange to bind.
  • exchange_destination – specifies the name of the destination exchange to bind
  • no_wait (bool) – if set, the server will not respond to the method
  • arguments (dict) – AMQP arguments to be passed when creating the exchange.
  • timeout (int) – wait for the server to respond after timeout