Pythonのカスタムイベントを作成する方法Asyncio

Question

MesosスケジューラMarathonには、非同期HTTP APIがあります。例えば。ユーザーが/v2/appsにJSONをポストしてアプリケーションをデプロイすると、デプロイメントIDが返されます。 idを使用してデプロイメント状態を/v2/deploymentsにポーリングするか、または/v2/eventsにサブスクライブしてdeployment_successイベントを探します。

コルーチンを使って非同期のPythonクライアントを作成したいと思います。例えば。 client.deploy_app(...)は、deployment_successイベントが到着しましたが、ブロックされていない状態に戻る必要があります。

どうすればこれらのメソッドをasyncioで実装できますか？イベントリスナーを作成するにはどうすればよいですか？イベントのループが作成されているように感じますが、イベントの登録方法はわかりません。あなたは（API docを参照）、ストリームを要求すべき展開の成功を追跡するために/v2/eventsを使用する場合は

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() 

：/v2/appsのために必要な非同期ポストHTTPリクエストを作成

Answer 1

はaiohttpモジュールで行うことができます。それはasynchronous iterationだとそれはaiohttpで達成することができます：あなただけの非同期例えば、コンテンツのライン・バイ・ラインあなたが必要とするイベントを待つをお読みください。

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を使用してください。この場合、機能はブロックされません。

import asyncio 
import aiohttp 


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


async def main(): 
    while True:  
     # Make request to see if deplayment finished: 
     res = await get('http://httpbin.org/get') 
     print(res['origin']) # break if ok here 
     # Async wait before next try: 
     await asyncio.sleep(3) 


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