Gestione degli errori nel codice asincrono

Quando si lavora con codice asincrono in Python utilizzando asyncio, la gestione degli errori richiede particolare attenzione perché le eccezioni possono propagarsi in modo diverso rispetto al codice sincrono. In un contesto sincrono, spesso si utilizza un blocco try-except per intercettare le eccezioni attorno al codice che potrebbe fallire. Questa stessa struttura funziona all'interno delle coroutine, ma ci sono ulteriori considerazioni quando si gestiscono task pianificati da asyncio.

Se viene sollevata un'eccezione all'interno di una coroutine async def e non viene intercettata, essa si propagherà fino al punto in cui la coroutine viene attesa. Se si utilizza await su una coroutine che solleva un'eccezione, l'eccezione verrà sollevata nel punto di attesa, consentendo di intercettarla con un blocco try-except. Tuttavia, se si pianifica una coroutine come task utilizzando asyncio.create_task() o asyncio.ensure_future(), l'eccezione non si propaga immediatamente. Invece, viene memorizzata nell'oggetto task e verrà sollevata solo quando si attende esplicitamente il task o si controlla il suo risultato.

Questo significa che, se non si gestiscono le eccezioni all'interno della coroutine o quando si attende il task, si potrebbero perdere errori critici, oppure le eccezioni potrebbero essere registrate dal ciclo degli eventi come "eccezioni non gestite". Per evitare ciò, è opportuno gestire le eccezioni all'interno della coroutine utilizzando try-except, oppure assicurarsi di attendere tutti i task e gestire correttamente i loro risultati. Questo approccio aiuta i programmi asincroni a rimanere stabili e facilita il debug.

Ecco un esempio di codice Python che mostra come gestire le eccezioni all'interno di una coroutine asincrona utilizzando un blocco try-except:

import asyncio

async def may_fail(n):
    try:
        if n % 2 == 0:
            raise ValueError(f"Even number error: {n}")
        print(f"Processed: {n}")
    except ValueError as e:
        print(f"Caught exception in coroutine: {e}")

async def main():
    tasks = [asyncio.create_task(may_fail(i)) for i in range(3)]
    await asyncio.gather(*tasks)

asyncio.run(main())

Output atteso:

Caught exception in coroutine: Even number error: 0
Processed: 1
Caught exception in coroutine: Even number error: 2

Questo esempio di codice definisce una coroutine asincrona may_fail che utilizza un blocco try-except per intercettare eccezioni di tipo ValueError quando viene elaborato un numero pari. Se n è pari, viene sollevata una ValueError e intercettata all'interno della coroutine, stampando un messaggio. La coroutine main pianifica tre task utilizzando asyncio.create_task() e li raccoglie con asyncio.gather(), assicurandosi che tutti i task vengano attesi. Questa configurazione garantisce che le eccezioni sollevate nelle coroutine vengano gestite correttamente e che nessuna eccezione non gestita rimanga nel ciclo degli eventi.