Welcome to aio-pika’s documentation!¶
aio_pika it’s a wrapper for the PIKA for asyncio and humans.
Features¶
- Completely asynchronous API.
- Object oriented API.
- 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).
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
Usage example¶
Some useful examples.
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"
async with connection:
# Creating channel
channel = await connection.channel()
# Declaring queue
queue = await channel.declare_queue(queue_name, auto_delete=True)
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)
async with connection:
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
)
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))
Tornado example¶
import asyncio
import tornado.ioloop
import tornado.web
from aio_pika import connect_robust, Message
tornado.ioloop.IOLoop.configure('tornado.platform.asyncio.AsyncIOLoop')
io_loop = tornado.ioloop.IOLoop.current()
asyncio.set_event_loop(io_loop.asyncio_loop)
QUEUE = asyncio.Queue()
class SubscriberHandler(tornado.web.RequestHandler):
async def get(self):
message = await QUEUE.get()
self.finish(message.body)
class PublisherHandler(tornado.web.RequestHandler):
async def post(self):
channel = await self.application.amqp_connection.channel()
await channel.default_exchange.publish(
Message(body=self.request.body),
routing_key="test",
)
await channel.close()
self.finish("OK")
def make_app():
return tornado.web.Application([
(r"/publish", PublisherHandler),
(r"/subscribe", SubscriberHandler),
])
async def initialize(app):
app.amqp_connection = await connect_robust()
channel = await app.amqp_connection.channel()
queue = await channel.declare_queue('test', auto_delete=True)
await queue.consume(QUEUE.put, no_ack=True)
if __name__ == "__main__":
app = make_app()
app.listen(8888)
io_loop.asyncio_loop.run_until_complete(initialize(app))
tornado.ioloop.IOLoop.current().start()
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]'
Tutorial¶
Patterns and helpers¶
Note
Available since aio-pika>=1.7.0
aio_pika includes some useful patterns for creating distributed systems.
Master/Worker¶
Helper which implements Master/Worker pattern. This applicable for balancing tasks between multiple workers.
The master creates tasks:
import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import Master
async def main():
connection = await connect_robust("amqp://guest:guest@127.0.0.1/")
# Creating channel
channel = await connection.channel()
master = Master(channel)
# Creates tasks by proxy object
for task_id in range(1000):
await master.proxy.my_task_name(task_id=task_id)
# Or using create_task method
for task_id in range(1000):
await master.create_task(
'my_task_name', kwargs=dict(task_id=task_id)
)
await connection.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Worker code:
import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import Master
async def worker(*, task_id):
print(task_id)
async def main():
connection = await connect_robust("amqp://guest:guest@127.0.0.1/")
# Creating channel
channel = await connection.channel()
master = Master(channel)
await master.create_worker('my_task_name', worker, auto_delete=True)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()
The one or multiple workers executes tasks.
RPC¶
Helper which implements Remote Procedure Call pattern. This applicable for balancing tasks between multiple workers.
The caller creates tasks and awaiting results:
import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import RPC
async def main():
connection = await connect_robust("amqp://guest:guest@127.0.0.1/")
# Creating channel
channel = await connection.channel()
rpc = await RPC.create(channel)
# Creates tasks by proxy object
for i in range(1000):
print(await rpc.proxy.multiply(x=100, y=i))
# Or using create_task method
for i in range(1000):
print(await rpc.call('multiply', kwargs=dict(x=100, y=i)))
await connection.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
One or multimple callees executing tasks:
import asyncio
from aio_pika import connect_robust
from aio_pika.patterns import RPC
async def multiply(*, x, y):
return x * y
async def main():
connection = await connect_robust("amqp://guest:guest@127.0.0.1/")
# Creating channel
channel = await connection.channel()
rpc = await RPC.create(channel)
await rpc.register('multiply', multiply, auto_delete=True)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()
Thanks for contributing¶
- @mosquito (author)
- @hellysmile (bug fixes and ideas)
- @alternativehood (bugfixes)
- @akhoronko
- @zyp
- @decaz
- @kajetanj
- @iselind
Versioning¶
This software follows Semantic Versioning