如何为Python Asyncio创建自定义事件

Question

Mesos调度程序Marathon具有异步HTTP API。例如。当通过向Qazxswpoi发布JSON来部署应用程序时，将返回部署ID。然后可以使用id在/v2/apps中轮询部署状态，或者通过订阅/v2/deployments并查找/v2/events事件。

我想用协同程序创建一个异步Python客户端。例如。 deployment_success应该在client.deploy_app(...)事件到来后返回，但不会阻止。

如何使用asyncio实现这些方法？如何创建事件监听器？感觉事件循环是为此而做的，但我不知道如何注册事件。

Answer 1

使用deployment_success模块可以创建/v2/apps所需的异步post http请求：

aiohttp

如果您想使用

import asyncio
import aiohttp


async def post(url, json):
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=json) as resp:
            return await resp.json()


async def main():
    res = await post('http://httpbin.org/post', {'test': 'object'})
    print(res['json'])  # {'test': 'object'}


loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
    loop.run_until_complete(loop.shutdown_asyncgens())
finally:
    loop.close()

来跟踪部署成功，您应该请求流（请参阅api /v2/events）。它可以在doc中用它的aiohttp来实现：你只需异步读取内容，逐行等待你需要的事件，例如：

asynchronous iteration

如果你想使用

import asyncio
import aiohttp


async def stream(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            async for line in resp.content:
                yield line


async def main():
    async for line in stream('http://httpbin.org/stream/10'):
        print(line)  # check if line contains event you need


loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
    loop.run_until_complete(loop.shutdown_asyncgens())
finally:
    loop.close()

，你应该使用/v2/deployments定期等待一些延迟。在这种情况下，您的功能不会阻塞：

asyncio.sleep