Skip to content
AMQP 0.9 client designed for asyncio and humans.
Python Makefile
Branch: master
Clone or download
Latest commit 6498575 Oct 30, 2019
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
aio_pika [fix] repr Oct 30, 2019
docs [fix] add client properties for examples Oct 16, 2019
tests [fix] repr Oct 30, 2019
.coafile [fix] improve setup.py and coala Jan 28, 2017
.coveragerc [fix] add newline to .coveragerc May 24, 2017
.drone.yml Fixes #226 (#227) Jun 28, 2019
.editorconfig [fix] keep 80 for examples Jan 5, 2018
.gitignore [fix] bump Apr 3, 2019
CHANGELOG.md [fix] bump 6.3.0 Oct 16, 2019
LICENSE Initial commit Sep 28, 2016
MANIFEST.in
Makefile [fix] upload target May 5, 2019
README.rst Fix documentation Sep 3, 2019
logo.svg add logo Apr 9, 2017
pylava.ini [fix] refactor .drone.yml Dec 11, 2018
pytest.ini drop python 3.4 support Sep 14, 2018
setup.py [fix] repr Oct 30, 2019
tox.ini [fea] add mypy Mar 25, 2019

README.rst

aio-pika

ReadTheDocs Coveralls Drone CI Latest Version

Wrapper for the aiormq for asyncio and humans.

Note

Since version 5.0.0 this library doesn't use pika as AMQP connector. Versions below 5.0.0 contains or requires pika's source codes.

See examples and the tutorial in documentation.

If you are newcomer in the RabbitMQ let's start the adopted official RabbitMQ tutorial

Features

  • Completely asynchronous API.
  • Object oriented API.
  • Transparent auto-reconnects with complete state recovery with connect_robust (e.g. declared queues or exchanges, consuming state and bindings).
  • Python 3.5+ compatible (include 3.7).
  • For python 3.4 users available aio-pika<4
  • Transparent publisher confirms support
  • Transactions support

Installation

pip install aio-pika

Usage example

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
    )

    async with connection:
        queue_name = "test_queue"

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

        # Declaring queue
        queue = await channel.declare_queue(
            queue_name,
            auto_delete=True
        )   # type: aio_pika.Queue

        async with queue.iterator() as queue_iter:
            # Cancel consuming after __aexit__
            async for message in queue_iter:
                async 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
    )

    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
    )

    await connection.close()


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
    await 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))

See another examples and the tutorial in documentation.

Versioning

This software follows Semantic Versioning

For contributors

Setting up development environment

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]'

Running Tests

NOTE: In order to run the tests locally you need to run a RabbitMQ instance with default user/password (guest/guest) and port (5672).

  • ProTip: Use Docker for this:
docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management

To test just run:

make test

Creating Pull Requests

You feel free to create pull request, but you should describe your cases and add some examples.

The changes should follow simple rules:

  • When your changes breaks public API you must increase the major version.
  • When your changes is safe for public API (e.g. added an argument with default value)
  • You have to add test cases (see tests/ folder)
  • You must add docstrings
  • You feel free to add yourself to "thank's to" section
You can’t perform that action at this time.