Tratamento de Erros em Código Assíncrono

Ao trabalhar com código assíncrono em Python utilizando asyncio, o tratamento de erros exige atenção especial, pois as exceções podem se propagar de maneira diferente em relação ao código síncrono. Em um contexto síncrono, normalmente se utiliza um bloco try-except para capturar exceções em torno de código que pode falhar. Essa mesma estrutura funciona dentro de corrotinas, mas há considerações adicionais ao lidar com tarefas agendadas pelo asyncio.

Se uma exceção for lançada dentro de uma corrotina async def e não for capturada, ela será propagada até o ponto em que a corrotina for aguardada. Se você utilizar await em uma corrotina que lança uma exceção, a exceção será lançada no local do await, permitindo capturá-la com um bloco try-except. No entanto, se você agendar uma corrotina como uma tarefa usando asyncio.create_task() ou asyncio.ensure_future(), a exceção não será propagada imediatamente. Em vez disso, ela é armazenada no objeto da tarefa e só será lançada quando você explicitamente aguardar a tarefa ou verificar seu resultado.

Isso significa que, se você não tratar as exceções dentro da corrotina ou ao aguardar a tarefa, pode acabar perdendo erros críticos, ou as exceções podem ser registradas pelo event loop como "exceções não tratadas". Para evitar isso, é recomendável tratar as exceções dentro da corrotina usando try-except, ou garantir que todas as tarefas sejam aguardadas e seus resultados tratados adequadamente. Essa abordagem ajuda a manter seus programas assíncronos estáveis e facilita a depuração.

Aqui está um exemplo de código Python que demonstra como tratar exceções dentro de uma corrotina assíncrona utilizando um bloco 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())

Saída esperada:

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

Este exemplo de código define uma corrotina assíncrona may_fail que utiliza um bloco try-except para capturar exceções ValueError ao processar um número par. Se n for par, uma ValueError é lançada e capturada dentro da corrotina, exibindo uma mensagem. A corrotina main agenda três tarefas usando asyncio.create_task() e as reúne com asyncio.gather(), garantindo que todas as tarefas sejam aguardadas. Essa configuração assegura que as exceções lançadas nas corrotinas sejam tratadas de forma adequada, evitando exceções não tratadas no event loop.