使用asyncio的python 3.5中的事件分派
accordian的Python项目详细描述
使用Asyncio的Python3.5中的事件分派
安装
pip install accordian
开始
import asyncio import accordian import random loop = asyncio.new_event_loop() dispatch = accordian.Dispatch(loop=loop) dispatch.register("my_event", ["id", "value"]) @dispatch.on("my_event") async def handle(id, value): sleep = 5.0 * random.random() print("Handling `my_event(id={})` in {} seconds.".format(id, sleep)) await asyncio.sleep(sleep, loop=loop) print("`Completed my_event(id={})`!".format(id)) ids = range(4) values = [random.random() for _ in ids] for id, value in zip(ids, values): params = {"id": id, "value": value} dispatch.trigger("my_event", params) loop.create_task(dispatch.start()) loop.run_until_complete(asyncio.sleep(0.01, loop=loop)) loop.run_until_complete(dispatch.stop())
贡献
欢迎投稿!请在提交PR之前确保tox
通过(包括flake8)。
开发
手风琴使用tox
、pytest
和flake8
。要设置所有设置:
# RECOMMENDED: create a virtualenv with: # mkvirtualenv accordian git clone https://github.com/numberoverzero/accordian.git pip install tox tox