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 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
Release history Release notifications
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Filename, size & hash SHA256 hash help | File type | Python version | Upload date |
---|---|---|---|
aio_pika-2.6.0-py3-none-any.whl (36.8 kB) Copy SHA256 hash SHA256 | Wheel | py3 | Apr 12, 2018 |
aio-pika-2.6.0.tar.gz (28.9 kB) Copy SHA256 hash SHA256 | Source | None | Apr 12, 2018 |