Wie füge ich Async-Generatoren in einen Python 3.5000-Generator ein?

Question

Ich habe Probleme beim Kombinieren von Async-Generatoren und tatsächlich laufen sie. Das liegt daran, dass ich sie nur durch eine Ereignisschleife gefunden habe, die einen iterierbaren und keinen Generator liefert. Lassen Sie mich das mit einem einfachen Beispiel illustrieren:

Nehmen wir an, ich habe eine Funktion google_search, die google scrappt (ich benutze die API nicht absichtlich). Es nimmt einen Suchstring auf und gibt einen Generator von Suchergebnissen zurück. Dieser Generator endet nicht, wenn die Seite zu Ende ist, die Funktion geht weiter zur nächsten Seite. Deshalb ist die GOOGLE_SEARCH Funktion einen möglicherweise fast endlos Generator liefert (es wird technisch immer am Ende aber oft kann man Millionen von Treffern für eine Suche auf Google erhalten)

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

Okay, so dass ich jetzt eine Funktion machen will, das mir erlaubt, um über mehrere google_search Generatoren zu iterieren. Ich würde so etwas wie dies mag:

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

Auf diese Weise habe ich eine einfache for-Schleife verwenden können google_searches um sich zu entspannen und meine Ergebnisse. Und der obige Code funktioniert gut, ist aber für eine ziemlich große Anzahl von Suchvorgängen sehr langsam. Der Code sendet eine Anfrage für die erste Suche, dann die zweite und so weiter, bis schließlich Ergebnisse erzielt werden. Ich möchte das (viel) beschleunigen. Meine erste Idee ist es, google_searches zu einer asynchronen Funktion zu machen (ich benutze python 3.6.3 und kann abwarten/async usw. verwenden). Dies erzeugt dann einen asynchronen Generator, der in Ordnung ist, aber ich kann ihn nur in einer anderen asynchronen Funktion oder einer Ereignisschleife ausführen. Und das Ausführen in einer Ereignisschleife mit run_until_complete (loop.gather (...)) gibt eine Liste von Ergebnissen anstelle eines normalen Generators zurück, was den Zweck zunichte macht, da wahrscheinlich zu viele Suchergebnisse in einer Liste enthalten sind.

Wie kann ich die google_searches-Funktion schneller machen (vorzugsweise async-Code verwenden, aber alles ist willkommen), indem Sie Anfragen asynchron ausführen, während Sie immer noch ein Vanilla-Generator sein müssen? Vielen Dank im Voraus!

Answer 1

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

Dies ist einfacher Synchrongenerator. Sie könnten requests darin verwenden, aber wenn Sie asynchrone aiohttp verwenden möchten, benötigen Sie asynchronous generator definiert mit async def.

Was zum Iterieren über mehrere asynchrone Generatoren kommt, ist interessanter. Sie können nicht einfach zip verwenden, da es mit einfachen Iterablen und nicht mit asynchronen Iterablen funktioniert. Sie sollten also Ihre eigenen implementieren (das würde auch das gleichzeitige Iterieren unterstützen).

ich machte einen kleinen Prototyp, ich denke, das macht, was Sie wollen:

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

Ausgang:

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 

Zeit zeigt, dass unterschiedliche Suchanfragen gleichzeitig ausgeführt werden.