Skip to main content

Wrapper for the PIKA for asyncio and humans.

Project description

aio-pika

ReadTheDocs 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

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

    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

Project details


Release history Release notifications | RSS feed

This version

4.7.0

Download files

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

Source Distribution

aio-pika-4.7.0.tar.gz (78.2 kB view details)

Uploaded Source

Built Distribution

aio_pika-4.7.0-py3-none-any.whl (90.2 kB view details)

Uploaded Python 3

File details

Details for the file aio-pika-4.7.0.tar.gz.

File metadata

  • Download URL: aio-pika-4.7.0.tar.gz
  • Upload date:
  • Size: 78.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/1.11.0 pkginfo/1.4.2 requests/2.20.1 setuptools/40.6.2 requests-toolbelt/0.8.0 tqdm/4.22.0 CPython/3.5.4

File hashes

Hashes for aio-pika-4.7.0.tar.gz
Algorithm Hash digest
SHA256 939b4168743daeba1484f34259abfc8687989daf2f39ad0fc59ecfe49b70129b
MD5 1b75ad62fd2c86a7524a31563a60bab3
BLAKE2b-256 a7ab039adef998e37a5cb6450cda7344b1d25c234e619be457d234a257bdc2bf

See more details on using hashes here.

File details

Details for the file aio_pika-4.7.0-py3-none-any.whl.

File metadata

  • Download URL: aio_pika-4.7.0-py3-none-any.whl
  • Upload date:
  • Size: 90.2 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/1.11.0 pkginfo/1.4.2 requests/2.20.1 setuptools/40.6.2 requests-toolbelt/0.8.0 tqdm/4.22.0 CPython/3.5.4

File hashes

Hashes for aio_pika-4.7.0-py3-none-any.whl
Algorithm Hash digest
SHA256 3353afdbb7c2c5e118e2be21381a47c68cfc40677a8b505c83f220f4ee41aaf1
MD5 287e24da9c771d31fd244b2fdc2c70f1
BLAKE2b-256 1077391c04ea6222a8160895aa4b91e8614544b1e39db0e90abca1865dade540

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page