Python 용 사용자 정의 이벤트 작성 방법 Asyncio

Question

메소 스 스케줄러 Marathon에는 비동기 HTTP API가 있습니다. 예 : 개발자가 /v2/apps에 JSON을 게시하여 앱을 배포하면 배포 ID가 반환됩니다. 그런 다음 ID를 사용하여 /v2/deployments에 배포 상태를 폴링하거나 /v2/events에 가입하여 deployment_success 이벤트를 찾습니다.

coroutines를 사용하여 비동기 Python 클라이언트를 만들고 싶습니다. 예 : client.deploy_app(...)은 deployment_success 이벤트가 도착했으나 차단되지 않은 경우 돌아와야합니다.

어떻게 이러한 메서드를 asyncio로 구현할 수 있습니까? 이벤트 수신기를 만들려면 어떻게해야합니까? 이벤트 루프가 만들어 졌다고 생각하지만 이벤트를 등록하는 방법을 알지 못합니다. /v2/apps에 필요한

Answer 1

만들기 비동기 포스트 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() 

당신이 스트림을 요청해야한다 배포 성공을 추적하는 /v2/events을 사용하려면 (API doc 참조). 이 asynchronous iteration의로 aiohttp 달성 할 수있다 : 당신은 그냥 비동기 당신이 필요로하는 이벤트를 기다리는 내용 한 줄 한 줄을 읽어, 예를 들면 : 당신이 /v2/deployments를 사용하려면

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

당신이 정기적으로 일부 지연 대기 요청해야합니다 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()