非同期ジェネレータをPython 3.5+のバニラジェネレータにマージする方法

Question

非同期ジェネレータを組み合わせて実際に実行する際に問題があります。これは、私がそれらを実行するために見つけた唯一の方法は、ジェネレータではなく、反復可能なものを返すイベントループによるものです。

GoogleがGoogleを検索するGoogleの検索機能（私は意図的にAPIを使用していない）を持っているとしましょう。検索文字列を取り込み、検索結果のジェネレータを返します。このジェネレータは、ページが終了しても終了しないので、次のページに進むことによって機能は継続します。したがってGOOGLE_SEARCH機能はオーケー、おそらくほぼ無限の発電機（それは技術的には常に終了しますが、多くの場合は、Googleの検索にヒット数百万を得ることができます）

def google_search(search_string): 
    # Basically uses requests/aiohttp and beautifulsoup 
    # to parse the resulting html and yield search results 
    # Assume this function works 
    ...... 

を返すので、今私は私を可能にする機能を作りたいです複数のgoogle_searchジェネレータを繰り返し処理します。

def google_searches(*search_strings): 
    for results in zip(google_search(query) for query in search_strings): 
     yield results 

このようにして、単純なforループを使用してgoogle_searchesを巻き戻して結果を得ることができます。そして、上記のコードはうまくいくが、かなりの数の検索では非常に遅い。コードは最初の検索の要求を送信し、次に2番目の検索の要求を送信して最終的に結果を返します。私はこれをスピードアップしたいと思っています（多く）。私の最初のアイデアは、非同期関数（私はPython 3.6.3を使用していて、await/asyncなどを使用できる）にgoogle_searchesを変更することです。これは、非同期ジェネレータを作成しますが、別の非同期関数またはイベントループでしか実行できません。また、run_until_complete（loop.gather（...））を使用してイベントループで実行すると、通常のジェネレータではなく結果のリストが返されます。リストに保持する検索結果が多すぎるため、目的を破棄します。

バニラジェネレータを使用しながらリクエストを非同期的に実行することによって、google_searches関数をより高速にする（できれば非同期コードを使用しますが何かが歓迎されます） ありがとうございます！

Answer 1

def google_search(search_string): 
    # Basically uses requests/aiohttp and beautifulsoup 

これは単純同期発電機です。 requestsを内部に使用できますが、非同期aiohttpを使用する場合は、asynchronous generatorをasync defと定義する必要があります。

複数の非同期ジェネレータを反復することは、より面白いです。 zipを使用することはできません。これは、非同期のiterablesではなくプレーンなiterableで動作するためです。したがって、独自に実装する必要があります（同時に反復処理もサポートされます）。

import asyncio 
import aiohttp 
import time 


# async versions of some builtins: 
async def anext(aiterator): 
    try: 
     return await aiterator.__anext__() 
    except StopAsyncIteration as exc: 
     raise exc 


def aiter(aiterable): 
    return aiterable.__aiter__() 


async def azip(*iterables): 
    iterators = [aiter(it) for it in iterables] 
    while iterators: 
     results = await asyncio.gather(
      *[anext(it) for it in iterators], 
      return_exceptions=True, 
     ) 
     yield tuple(results) 


# emulating grabbing: 
async def request(url): 
    async with aiohttp.ClientSession() as session: 
     async with session.get(url) as resp: 
      return await resp.text() 


async def google_search(search_string): 
    for i in range(999): # big async generator 
     url = 'http://httpbin.org/delay/{}'.format(i) # increase delay to better see concurency 
     j = await request(url) 
     yield search_string + ' ' + str(i) 


async def google_searches(*search_strings): 
    async for results in azip(*[google_search(s) for s in search_strings]): 
     for result in results: 
      yield result 


# test it works: 
async def main(): 
    async for result in google_searches('first', 'second', 'third'): 
     print(result, int(time.time())) 


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

出力：

first 0 1514759561 
second 0 1514759561 
third 0 1514759561 
first 1 1514759562 
second 1 1514759562 
third 1 1514759562 
first 2 1514759564 
second 2 1514759564 
third 2 1514759564 
first 3 1514759567 
second 3 1514759567 
third 3 1514759567 

時間が異なる検索を同時に実行することを示している

は、私が何をしたいんだと思う少し試作品を作りました。