Skip to main content
This is a pre-production deployment of Warehouse. Changes made here affect the production instance of PyPI (pypi.python.org).
Help us improve Python packaging - Donate today!

Wrapper for the PIKA for asyncio and humans.

Project Description

aio-pika

Wrapper for the PIKA for asyncio and humans. 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.4+ compatible (include 3.6).
  • 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)

    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 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 receiver:

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

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
Release History

Release History

This version
History Node

1.4.2

History Node

1.4.1

History Node

1.4.0

History Node

1.3.5

History Node

1.3.4

History Node

1.3.3

History Node

1.3.2

History Node

1.3.1

History Node

1.3.0

History Node

1.2.0

History Node

1.1.1

History Node

1.1.0

History Node

1.0.1

History Node

1.0.0

History Node

0.21.0

History Node

0.20.1

History Node

0.20.0

History Node

0.19.0

History Node

0.18.0

History Node

0.17.0

History Node

0.16.0

History Node

0.15.1

History Node

0.15.0

History Node

0.14.1

History Node

0.14.0

History Node

0.12.3

History Node

0.12.2

History Node

0.12.1

History Node

0.12.0

History Node

0.11.1

History Node

0.11.0

History Node

0.10.7

History Node

0.10.6

History Node

0.10.5

History Node

0.10.4

History Node

0.10.2

History Node

0.10.1

History Node

0.10.0

History Node

0.9.7

History Node

0.9.5

History Node

0.9.4

History Node

0.9.3

History Node

0.9.2

History Node

0.9.1

History Node

0.9.0

History Node

0.6.4

History Node

0.6.3

History Node

0.6.2

History Node

0.6.1

History Node

0.6.0

History Node

0.5.7

History Node

0.5.6

History Node

0.5.5

History Node

0.5.4

History Node

0.5.3

History Node

0.5.2

History Node

0.5.1

History Node

0.5.0

Download Files

Download Files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
aio_pika-1.4.2-py3-none-any.whl (32.7 kB) Copy SHA256 Checksum SHA256 3.5 Wheel Sep 25, 2017
aio-pika-1.4.2.tar.gz (22.6 kB) Copy SHA256 Checksum SHA256 Source Sep 25, 2017

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting