Welcome to aio-pika’s documentation!#
aio-pika is a wrapper for the aiormq for asyncio and humans.
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.6+ compatible.
For python 3.5 users available aio-pika<7
Transparent publisher confirms support
Transactions support
Completely type-hints coverage.
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
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]'
Table Of Contents#
- Quick start
- Patterns and helpers
- RabbitMQ tutorial
- API Reference
Thanks for contributing#
@mosquito (author)
@decaz (steel persuasiveness while code review)
@heckad (bug fixes)
@smagafurov (bug fixes)
@hellysmile (bug fixes and ideas)
@altvod (bug fixes)
@alternativehood (bugfixes)
@cprieto (bug fixes)
@akhoronko (bug fixes)
@iselind (bug fixes)
@DXist (bug fixes)
@blazewicz (bug fixes)
@chibby0ne (bug fixes)
@jmccarrell (bug fixes)
@taybin (bug fixes)
@ollamh (bug fixes)
@DriverX (bug fixes)
@brianmedigate (bug fixes)
@dan-stone (bug fixes)
@Kludex (bug fixes)
@bmario (bug fixes)
@tzoiker (bug fixes)
@Pehat (bug fixes)
@WindowGenerator (bug fixes)
@dhontecillas (bug fixes)
@tilsche (bug fixes)
@leenr (bug fixes)
@la0rg (bug fixes)
@SolovyovAlexander (bug fixes)
@kremius (bug fixes)
@zyp (bug fixes)
@kajetanj (bug fixes)
@Alviner (moral support, debug sessions and good mood)
@Pavkazzz (composure, and patience while debug sessions)
@bbrodriges (supplying grammar while writing documentation)
@dizballanze (review, grammar)
See also#
aiormq#
aiormq is a pure python AMQP client library. It is under the hood of aio-pika and might to be used when you really loving works with the protocol low level. Following examples demonstrates the user API.
Simple consumer:
import asyncio
import aiormq
async def on_message(message):
"""
on_message doesn't necessarily have to be defined as async.
Here it is to show that it's possible.
"""
print(f" [x] Received message {message!r}")
print(f"Message body is: {message.body!r}")
print("Before sleep!")
await asyncio.sleep(5) # Represents async I/O operations
print("After sleep!")
async def main():
# Perform connection
connection = await aiormq.connect("amqp://guest:guest@localhost/")
# Creating a channel
channel = await connection.channel()
# Declaring queue
declare_ok = await channel.queue_declare('helo')
consume_ok = await channel.basic_consume(
declare_ok.queue, on_message, no_ack=True
)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()
Simple publisher:
import asyncio
from typing import Optional
import aiormq
from aiormq.abc import DeliveredMessage
MESSAGE: Optional[DeliveredMessage] = None
async def main():
global MESSAGE
body = b'Hello World!'
# Perform connection
connection = await aiormq.connect("amqp://guest:guest@localhost//")
# Creating a channel
channel = await connection.channel()
declare_ok = await channel.queue_declare("hello", auto_delete=True)
# Sending the message
await channel.basic_publish(body, routing_key='hello')
print(f" [x] Sent {body}")
MESSAGE = await channel.basic_get(declare_ok.queue)
print(f" [x] Received message from {declare_ok.queue!r}")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
assert MESSAGE is not None
assert MESSAGE.routing_key == "hello"
assert MESSAGE.body == b'Hello World!'
The patio and the patio-rabbitmq#
PATIO is an acronym for Python Asynchronous Tasks for AsyncIO - an easily extensible library, for distributed task execution, like celery, only targeting asyncio as the main design approach.
patio-rabbitmq provides you with the ability to use RPC over RabbitMQ services with extremely simple implementation:
from patio import Registry, ThreadPoolExecutor
from patio_rabbitmq import RabbitMQBroker
rpc = Registry(project="patio-rabbitmq", auto_naming=False)
@rpc("sum")
def sum(*args):
return sum(args)
async def main():
async with ThreadPoolExecutor(rpc, max_workers=16) as executor:
async with RabbitMQBroker(
executor, amqp_url="amqp://guest:guest@localhost/",
) as broker:
await broker.join()
And the caller side might be written like this:
import asyncio
from patio import NullExecutor, Registry
from patio_rabbitmq import RabbitMQBroker
async def main():
async with NullExecutor(Registry(project="patio-rabbitmq")) as executor:
async with RabbitMQBroker(
executor, amqp_url="amqp://guest:guest@localhost/",
) as broker:
print(await asyncio.gather(
*[
broker.call("mul", i, i, timeout=1) for i in range(10)
]
))
Propan#
Propan is a powerful and easy-to-use Python framework for building event-driven applications that interact with any MQ Broker.
If you need no deep dive into RabbitMQ details, you can use more high-level Propan interfaces:
from propan import PropanApp, RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
@broker.handle("user")
async def user_created(user_id: int):
assert isinstance(user_id, int)
return f"user-{user_id}: created"
@app.after_startup
async def pub_smth():
assert (
await broker.publish(1, "user", callback=True)
) == "user-1: created"
Also, Propan validates messages by pydantic, generates your project AsyncAPI spec, tests application locally, RPC calls, and more.
In fact, it is a high-level wrapper on top of aio-pika, so you can use both of these libraries’ advantages at the same time.
python-socketio#
Socket.IO is a transport protocol that enables real-time bidirectional event-based communication between clients (typically, though not always, web browsers) and a server. This package provides Python implementations of both, each with standard and asyncio variants.
Also this package is suitable for building messaging services over RabbitMQ via aio-pika adapter:
import socketio
from aiohttp import web
sio = socketio.AsyncServer(client_manager=socketio.AsyncAioPikaManager())
app = web.Application()
sio.attach(app)
@sio.event
async def chat_message(sid, data):
print("message ", data)
if __name__ == '__main__':
web.run_app(app)
And a client is able to call chat_message the following way:
import asyncio
import socketio
sio = socketio.AsyncClient()
async def main():
await sio.connect('http://localhost:8080')
await sio.emit('chat_message', {'response': 'my response'})
if __name__ == '__main__':
asyncio.run(main())
The taskiq and the taskiq-aio-pika#
Taskiq is an asynchronous distributed task queue for python. The project takes inspiration from big projects such as Celery and Dramatiq. But taskiq can send and run both the sync and async functions.
The library provides you with aio-pika broker for running tasks too.
from taskiq_aio_pika import AioPikaBroker
broker = AioPikaBroker()
@broker.task
async def test() -> None:
print("nothing")
async def main():
await broker.startup()
await test.kiq()
Rasa#
With over 25 million downloads, Rasa Open Source is the most popular open source framework for building chat and voice-based AI assistants.
With Rasa, you can build contextual assistants on:
Facebook Messenger
Slack
Google Hangouts
Webex Teams
Microsoft Bot Framework
Rocket.Chat
Mattermost
Telegram
Twilio
Your own custom conversational channels or voice assistants as:
Alexa Skills
Google Home Actions
Rasa helps you build contextual assistants capable of having layered conversations with lots of back-and-forth. In order for a human to have a meaningful exchange with a contextual assistant, the assistant needs to be able to use context to build on things that were previously discussed – Rasa enables you to build assistants that can do this in a scalable way.
And it also uses aio-pika to interact with RabbitMQ deep inside!
Versioning#
This software follows Semantic Versioning