Создайте генератор, который дает результаты coroutine при завершении сопрограммы

Question

В настоящее время у меня есть неэффективный синхронный генератор, который последовательно выполняет множество HTTP-запросов и дает результаты. Я хотел бы использовать asyncio и aiohttp для параллелизации запросов и тем самым ускорить этот генератор, но я хочу сохранить его как обычный генератор (а не PEP 525 async generator), так что неасинхронный код, который его вызывает, не нужен быть изменен. Как я могу создать такой генератор?

Answer 1

asyncio.as_completed(), в настоящее время едва документированный, принимает итерабельность сопрограмм или фьючерсов и возвращает итеративный фьючерс в порядке завершения ввода фьючерсов. Обычно, вы бы петлю на его результат и await членов из внутри async функции ...

import asyncio 

async def first(): 
    await asyncio.sleep(5) 
    return 'first' 

async def second(): 
    await asyncio.sleep(1) 
    return 'second' 

async def third(): 
    await asyncio.sleep(3) 
    return 'third' 

async def main(): 
    for future in asyncio.as_completed([first(), second(), third()]): 
     print(await future) 

loop = asyncio.get_event_loop() 

# Prints 'second', then 'third', then 'first' 
loop.run_until_complete(main()) 

... но для целей данного вопроса, что мы хотим, чтобы быть в состоянии дать эти результаты от обычного генератора, так что нормальный синхронный код может потреблять их, даже не зная, что под капотом используются функции async. Мы можем сделать это по телефону loop.run_until_complete() на фьючерсах, обеспечиваемых наших as_completed вызова ...

import asyncio 

async def first(): 
    await asyncio.sleep(5) 
    return 'first' 

async def second(): 
    await asyncio.sleep(1) 
    return 'second' 

async def third(): 
    await asyncio.sleep(3) 
    return 'third' 

def ordinary_generator(): 
    loop = asyncio.get_event_loop() 
    for future in asyncio.as_completed([first(), second(), third()]): 
     yield loop.run_until_complete(future) 

# Prints 'second', then 'third', then 'first' 
for element in ordinary_generator(): 
    print(element) 

Таким образом, мы подвержены наша асинхронную код, не включенные в асинхронном-земли таким образом, что не требует чтобы определить любые функции как async, или даже знать, что ordinary_generator использует asyncio под капотом.

В качестве альтернативы реализации ordinary_generator(), который предлагает большую гибкость в некоторых случаях, мы можем неоднократно называть asyncio.wait() с FIRST_COMPLETED флагом вместо зацикливания над as_completed():

import concurrent.futures 

def ordinary_generator(): 
    loop = asyncio.get_event_loop() 
    pending = [first(), second(), third()] 
    while pending: 
     done, pending = loop.run_until_complete(
      asyncio.wait(
       pending, 
       return_when=concurrent.futures.FIRST_COMPLETED 
      ) 
     ) 
     for job in done: 
      yield job.result() 

Этот подход, сохраняя список pending рабочих мест, имеет то преимущество, что мы можем адаптировать его для добавления заданий в список pending на лету. Это полезно в случаях использования, когда наши асинхронные задания могут добавить непредсказуемое количество дополнительных заданий в очередь - например, паутину, которая следует за всеми ссылками на каждой посещаемой странице.