Gestione asincrona delle eccezioni in Python

Question

Ho il seguente codice utilizzando asyncio e aiohttp per effettuare richieste HTTP asincrone.

import sys 
import asyncio 
import aiohttp 

@asyncio.coroutine 
def get(url): 
    try: 
     print('GET %s' % url) 
     resp = yield from aiohttp.request('GET', url) 
    except Exception as e: 
     raise Exception("%s has error '%s'" % (url, e)) 
    else: 
     if resp.status >= 400: 
      raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason)) 

    return (yield from resp.text()) 

@asyncio.coroutine 
def fill_data(run): 
    url = 'http://www.google.com/%s' % run['name'] 
    run['data'] = yield from get(url) 

def get_runs(): 
    runs = [ {'name': 'one'}, {'name': 'two'} ] 
    loop = asyncio.get_event_loop() 
    task = asyncio.wait([fill_data(r) for r in runs]) 
    loop.run_until_complete(task) 
    return runs 

try: 
    get_runs() 
except Exception as e: 
    print(repr(e)) 
    sys.exit(1) 

Per qualche ragione, le eccezioni sollevate all'interno della funzione get non sono catturati:

Future/Task exception was never retrieved 
Traceback (most recent call last): 
    File "site-packages/asyncio/tasks.py", line 236, in _step 
    result = coro.send(value) 
    File "mwe.py", line 25, in fill_data 
    run['data'] = yield from get(url) 
    File "mwe.py", line 17, in get 
    raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason)) 
Exception: http://www.google.com/two has error '404: Not Found' 

Quindi, qual è corretto modo di gestire le eccezioni sollevate da couroutines?

Answer 1

asyncio.wait in realtà non consumare il Futures passati ad esso, si aspetta solo per loro di completare, e poi restituisce il Future oggetti:

coroutineasyncio.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

Attendi gli oggetti Futures e coroutine dati dalla sequenza di future da completare. Le Coroutine verranno incapsulate nello in Attività. Restituisce due serie di Future: (terminato, in sospeso).

fino a quando effettivamente yield from gli elementi nell'elenco done, che rimarranno non consumato. Dal momento che il tuo programma esce senza consumare i futuri, vedi i messaggi di "eccezione non è mai stato recuperato".

Per il vostro caso d'uso, probabilmente ha più senso usare asyncio.gather, che effettivamente consumare ogni Future, e quindi restituire un singolo Future che aggrega tutti i loro risultati (o solleva il primo Exception lanciata da un futuro in ingresso elenco).

def get_runs(): 
    runs = [ {'name': 'one'}, {'name': 'two'} ] 
    loop = asyncio.get_event_loop() 
    tasks = asyncio.gather(*[fill_data(r) for r in runs]) 
    loop.run_until_complete(tasks) 
    return runs 

uscita:

GET http://www.google.com/two 
GET http://www.google.com/one 
Exception("http://www.google.com/one has error '404: Not Found'",) 

Nota che asyncio.gather consente in realtà di personalizzare il suo comportamento quando uno dei futuri solleva un'eccezione; il comportamento predefinito è quello di aumentare la prima eccezione che colpisce, ma può anche semplicemente tornare ogni oggetto eccezione nella lista di uscita:

asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

Return un futuro aggregando i risultati dalla data coroutine oggetti o futuri.

Tutti i futures devono condividere lo stesso ciclo di eventi. Se tutte le attività vengono eseguite con successo , il risultato del risultato restituito è l'elenco dei risultati (in l'ordine della sequenza originale, non necessariamente l'ordine di arrivo dei risultati ).Se return_exceptions è True, le eccezioni nelle attività sono considerate uguali ai risultati riusciti e raccolte nell'elenco dei risultati ; in caso contrario, la prima eccezione sollevata sarà immediatamente propagata al futuro restituito.

Answer 2

asyncio fornisce un'API per la gestione degli errori personalizzati, fare riferimento al documento https://docs.python.org/3/library/asyncio-eventloop.html#error-handling-api

Answer 3

Per eseguire il debug o "gestire" le eccezioni in callback:

coroutine che restituiscono qualche risultato o alzare eccezioni:

@asyncio.coroutine 
def async_something_entry_point(self): 
    try: 
     return self.real_stuff_which_throw_exceptions() 
    except: 
     raise Exception(some_identifier_here + ' ' + traceback.format_exc()) 

E callback:

def callback(self, future: asyncio.Future): 
    exc = future.exception() 
    if exc: 
     # Handle wonderful empty TimeoutError exception 
     if type(exc) == TimeoutError: 
      self.logger('<Some id here> callback exception TimeoutError') 
     else: 
      self.logger("<Some id here> callback exception " + str(exc)) 

    # store your result where you want 
    self.result.append(
     future.result() 
    )