Welcome to aio-pika’s documentation!

Coveralls Travis CI Latest Version https://img.shields.io/pypi/wheel/aio-pika.svg https://img.shields.io/pypi/pyversions/aio-pika.svg https://img.shields.io/pypi/l/aio-pika.svg

aio_pika it’s a wrapper for the PIKA for asyncio and humans.

Features

  • Completely asynchronous API.
  • Object oriented API.
  • Auto-reconnects with complete state recovery with connect_robust (e.g. declared queues or exchanges, consuming state and bindings).
  • Python 3.4+ compatible (include 3.6).

Installation

Installation with pip:

pip install aio-pika

Installation from git:

# via pip
pip install https://github.com/mosquito/aio-pika/archive/master.zip

# manually
git clone https://github.com/mosquito/aio-pika.git
cd aio-pika
python setup.py install

Usage example

Some useful examples.

Simple consumer

import asyncio
import aio_pika


async def main(loop):
    connection = await aio_pika.connect_robust(
        "amqp://guest:guest@127.0.0.1/", loop=loop
    )

    queue_name = "test_queue"

    async with connection:
        # Creating channel
        channel = await connection.channel()

        # Declaring queue
        queue = await channel.declare_queue(queue_name, auto_delete=True)

        async for message in queue:
            with message.process():
                print(message.body)

                if queue.name in message.body.decode():
                    break


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

Simple publisher

import asyncio
import aio_pika


async def main(loop):
    connection = await aio_pika.connect_robust("amqp://guest:guest@127.0.0.1/", loop=loop)

    async with connection:
        routing_key = "test_queue"

        channel = await connection.channel()    # type: aio_pika.Channel

        await channel.default_exchange.publish(
            aio_pika.Message(
                body='Hello {}'.format(routing_key).encode()
            ),
            routing_key=routing_key
        )


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

Get single message example

import asyncio
from aio_pika import connect_robust, Message


async def main(loop):
    connection = await connect_robust("amqp://guest:guest@127.0.0.1/", loop=loop)

    queue_name = "test_queue"
    routing_key = "test_queue"

    # Creating channel
    channel = await connection.channel()

    # Declaring exchange
    exchange = await channel.declare_exchange('direct', auto_delete=True)

    # Declaring queue
    queue = await channel.declare_queue(queue_name, auto_delete=True)

    # Binding queue
    await queue.bind(exchange, routing_key)

    await exchange.publish(
        Message(
            bytes('Hello', 'utf-8'),
            content_type='text/plain',
            headers={'foo': 'bar'}
        ),
        routing_key
    )

    # Receiving message
    incoming_message = await queue.get(timeout=5)

    # Confirm message
    incoming_message.ack()

    await queue.unbind(exchange, routing_key)
    await queue.delete()
    await connection.close()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

Tornado example

import asyncio
import tornado.ioloop
import tornado.web

from aio_pika import connect_robust, Message

tornado.ioloop.IOLoop.configure('tornado.platform.asyncio.AsyncIOLoop')
io_loop = tornado.ioloop.IOLoop.current()
asyncio.set_event_loop(io_loop.asyncio_loop)


QUEUE = asyncio.Queue()


class SubscriberHandler(tornado.web.RequestHandler):
    async def get(self):
        message = await QUEUE.get()
        self.finish(message.body)


class PublisherHandler(tornado.web.RequestHandler):
    async def post(self):
        channel = await self.application.amqp_connection.channel()
        await channel.default_exchange.publish(
            Message(body=self.request.body),
            routing_key="test",
        )

        await channel.close()
        self.finish("OK")


def make_app():
    return tornado.web.Application([
        (r"/publish", PublisherHandler),
        (r"/subscribe", SubscriberHandler),
    ])


async def initialize(app):
    app.amqp_connection = await connect_robust()

    channel = await app.amqp_connection.channel()
    queue = await channel.declare_queue('test', auto_delete=True)
    await queue.consume(QUEUE.put, no_ack=True)


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)

    io_loop.asyncio_loop.run_until_complete(initialize(app))
    tornado.ioloop.IOLoop.current().start()

Development

Clone the project:

git clone https://github.com/mosquito/aio-pika.git
cd aio-pika

Create a new virtualenv for aio_pika:

virtualenv -p python3.5 env

Install all requirements for aio_pika:

env/bin/pip install -e '.[develop]'

Patterns and helpers

Note

Available since aio-pika>=1.7.0

aio_pika includes some useful patterns for creating distributed systems.

Master/Worker

Helper which implements Master/Worker pattern. This applicable for balancing tasks between multiple workers.

The master creates tasks:

import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import Master


async def main():
    connection = await connect_robust("amqp://guest:guest@127.0.0.1/")

    # Creating channel
    channel = await connection.channel()

    master = Master(channel)

    # Creates tasks by proxy object
    for task_id in range(1000):
        await master.proxy.my_task_name(task_id=task_id)

    # Or using create_task method
    for task_id in range(1000):
        await master.create_task(
            'my_task_name', kwargs=dict(task_id=task_id)
        )

    await connection.close()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Worker code:

import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import Master


async def worker(*, task_id):
    print(task_id)


async def main():
    connection = await connect_robust("amqp://guest:guest@127.0.0.1/")

    # Creating channel
    channel = await connection.channel()

    master = Master(channel)
    await master.create_worker('my_task_name', worker, auto_delete=True)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.create_task(main())
    loop.run_forever()

The one or multiple workers executes tasks.

RPC

Helper which implements Remote Procedure Call pattern. This applicable for balancing tasks between multiple workers.

The caller creates tasks and awaiting results:

import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import RPC


async def main():
    connection = await connect_robust("amqp://guest:guest@127.0.0.1/")

    # Creating channel
    channel = await connection.channel()

    rpc = await RPC.create(channel)

    # Creates tasks by proxy object
    for i in range(1000):
        print(await rpc.proxy.multiply(x=100, y=i))

    # Or using create_task method
    for i in range(1000):
        print(await rpc.call('multiply', kwargs=dict(x=100, y=i)))

    await connection.close()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

One or multimple callees executing tasks:

import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import RPC


async def multiply(*, x, y):
    return x * y


async def main():
    connection = await connect_robust("amqp://guest:guest@127.0.0.1/")

    # Creating channel
    channel = await connection.channel()

    rpc = await RPC.create(channel)
    await rpc.register('multiply', multiply, auto_delete=True)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.create_task(main())
    loop.run_forever()

Thanks for contributing

Versioning

This software follows Semantic Versioning