Bem-vindo à aula 65 do curso de Python! Até aqui, você já deve ter compreendido os fundamentos de programação assíncrona com async e await. Agora é hora de colocar a mão na massa e explorar as ferramentas mais poderosas do módulo asyncio para escrever código concorrente eficiente e legível. Nesta aula, você vai aprender a gerenciar múltiplas tarefas, combinar corrotinas, definir limites de tempo e realizar operações de I/O assíncrono de forma prática.

Vamos mergulhar em exemplos reais e ver como o asyncio pode transformar a forma como você lida com operações que dependem de espera, como chamadas de rede, leitura de arquivos e consultas a bancos de dados. Ao final, você terá uma base sólida para construir aplicações Python escaláveis e responsivas.

Tarefas (create_task)

Quando você define uma corrotina (função async def), ela não é executada até que seja agendada no event loop. Uma das maneiras mais comuns de agendar uma corrotina é usando asyncio.create_task(). Essa função cria uma tarefa (Task) que agenda a execução da corrotina de forma concorrente com outras tarefas, permitindo que você não precise esperar uma terminar para iniciar outra.

O create_task é essencial quando você deseja executar várias corrotinas simultaneamente, sem depender de uma delas para iniciar as demais. A tarefa retornada pode ser aguardada com await para obter o resultado, ou você pode deixá-la rodando em segundo plano enquanto executa outras operações.

import asyncio

async def tarefa(nome, tempo):
    print(f"Iniciando {nome}")
    await asyncio.sleep(tempo)
    print(f"Finalizando {nome}")
    return f"Resultado de {nome}"

async def main():
    # Criando tarefas a partir de corrotinas
    t1 = asyncio.create_task(tarefa("Tarefa 1", 2))
    t2 = asyncio.create_task(tarefa("Tarefa 2", 1))

    # Aguardando ambas as tarefas
    resultado1 = await t1
    resultado2 = await t2

    print(resultado1)
    print(resultado2)

asyncio.run(main())

Observe que as tarefas são criadas antes de qualquer await que as aguarde. Isso permite que ambas comecem a executar imediatamente, e o event loop alterna entre elas durante os períodos de espera (como o sleep). Esse é o princípio fundamental da concorrência assíncrona.

gather

O asyncio.gather() é uma função de conveniência que permite aguardar várias corrotinas ou tarefas de uma só vez, retornando uma lista com os resultados na mesma ordem. É uma forma elegante de paralelizar operações independentes, sem precisar gerenciar tarefas manualmente.

O gather é especialmente útil quando você deseja executar um conjunto fixo de corrotinas e coletar todos os resultados. Ele também pode tratar exceções de forma agregada, dependendo do parâmetro return_exceptions.

import asyncio

async def operacao(id, tempo):
    await asyncio.sleep(tempo)
    return f"Operação {id} concluída"

async def main():
    resultados = await asyncio.gather(
        operacao(1, 2),
        operacao(2, 1),
        operacao(3, 3)
    )
    print(resultados)

asyncio.run(main())

Se uma das corrotinas lançar uma exceção, o gather propaga a exceção para o chamador, a menos que você defina return_exceptions=True, caso em que a exceção é retornada como parte da lista de resultados. Isso oferece flexibilidade para tratar erros de forma controlada.

import asyncio

async def falhar():
    raise ValueError("Erro proposital")

async def ok():
    return "OK"

async def main():
    resultados = await asyncio.gather(
        ok(),
        falhar(),
        return_exceptions=True
    )
    for r in resultados:
        if isinstance(r, Exception):
            print(f"Erro capturado: {r}")
        else:
            print(r)

asyncio.run(main())

Timeouts

Em aplicações reais, é comum precisar limitar o tempo de espera de uma operação assíncrona. O asyncio oferece duas formas principais de aplicar timeouts: asyncio.wait_for() e asyncio.timeout() (disponível a partir do Python 3.11).

O wait_for aguarda uma corrotina ou tarefa com um tempo máximo; se o tempo exceder, a corrotina é cancelada e uma exceção TimeoutError é lançada. Já o timeout é um gerenciador de contexto que pode ser usado para envolver um bloco de código, cancelando a operação em andamento se o tempo limite for excedido.

import asyncio

async def operacao_demorada():
    await asyncio.sleep(5)
    return "Terminou"

async def main():
    try:
        # Timeout de 2 segundos
        resultado = await asyncio.wait_for(operacao_demorada(), timeout=2)
        print(resultado)
    except asyncio.TimeoutError:
        print("A operação excedeu o tempo limite!")

asyncio.run(main())

Com o timeout do Python 3.11+, você pode usar a sintaxe de contexto:

import asyncio

async def main():
    try:
        async with asyncio.timeout(2):
            await asyncio.sleep(5)
    except TimeoutError:
        print("Tempo esgotado!")

asyncio.run(main())

É importante notar que, ao ocorrer timeout, a tarefa cancelada recebe uma exceção asyncio.CancelledError internamente, que pode ser tratada para realizar limpezas, como fechar conexões.

I/O assíncrono

O asyncio é particularmente eficaz para operações de I/O, como leitura/escrita em sockets, arquivos e comunicação com serviços web. Para isso, o Python oferece bibliotecas assíncronas como aiohttp para HTTP, aiofiles para arquivos e aiomysql para MySQL. Nesta seção, veremos um exemplo prático de I/O assíncrono usando aiohttp para fazer múltiplas requisições HTTP concorrentes.

Antes de executar, instale a biblioteca com pip install aiohttp. O exemplo abaixo busca várias URLs simultaneamente e coleta o tamanho do conteúdo retornado.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        content = await response.text()
        return len(content)

async def main():
    urls = [
        "https://example.com",
        "https://httpbin.org/get",
        "https://jsonplaceholder.typicode.com/todos/1"
    ]
    async with aiohttp.ClientSession() as session:
        tarefas = [fetch(session, url) for url in urls]
        tamanhos = await asyncio.gather(*tarefas)
        for url, tamanho in zip(urls, tamanhos):
            print(f"{url}: {tamanho} bytes")

asyncio.run(main())

Observe como o gather é usado para executar as requisições em paralelo, aproveitando ao máximo o tempo de espera de rede. Esse padrão é escalável para centenas de requisições simultâneas, tornando seu código muito mais eficiente do que abordagens sequenciais.

Boas práticas e observações finais

Ao trabalhar com asyncio, lembre-se de que ele não é um substituto para multithreading em tarefas intensivas de CPU. Para operações de CPU-bound, considere usar concurrent.futures ou multiprocessamento. Além disso, sempre que possível, evite misturar código bloqueante (síncrono) dentro de corrotinas, pois isso pode travar o event loop. Use bibliotecas assíncronas específicas ou execute código bloqueante em threads com asyncio.to_thread().

Outra prática recomendada é usar asyncio.run() como ponto de entrada, garantindo que o event loop seja criado e fechado corretamente. Em aplicações maiores, considere usar bibliotecas como AnyIO ou Trio para compatibilidade entre backends.

Referências

Exercícios

  1. Crie uma corrotina que receba um nome e um tempo, aguarde o tempo e imprima "Olá, {nome}!" após a espera. Use create_task para executar três dessas corrotinas concorrentemente no main().

    ✓ Resposta:
    import asyncio

async def saudacao(nome, tempo):
    await asyncio.sleep(tempo)
    print(f"Olá, {nome}!")

async def main():
    tasks = [
        asyncio.create_task(saudacao("Alice", 2)),
        asyncio.create_task(saudacao("Bob", 1)),
        asyncio.create_task(saudacao("Carol", 3))
    ]
    await asyncio.gather(*tasks)

asyncio.run(main())

  2. Escreva um programa que use asyncio.gather() para executar três corrotinas que retornam números após um atraso. Some os resultados e imprima a soma.

    ✓ Resposta:
    import asyncio

async def numero(valor, atraso):
    await asyncio.sleep(atraso)
    return valor

async def main():
    resultados = await asyncio.gather(
        numero(10, 1),
        numero(20, 2),
        numero(30, 3)
    )
    soma = sum(resultados)
    print(f"Soma: {soma}")

asyncio.run(main())

  3. Implemente uma corrotina que simule uma operação que pode demorar, mas deve ser limitada a 3 segundos. Use asyncio.wait_for para lançar um timeout se exceder. Teste com um atraso de 5 segundos e capture a exceção.

    ✓ Resposta:
    import asyncio

async def operacao_lenta():
    await asyncio.sleep(5)
    return "Concluído"

async def main():
    try:
        resultado = await asyncio.wait_for(operacao_lenta(), timeout=3)
        print(resultado)
    except asyncio.TimeoutError:
        print("Tempo excedido!")

asyncio.run(main())

  4. Usando aiohttp, escreva um programa que faça requisições GET para três URLs diferentes e imprima o status code de cada uma. Use gather para executar concorrentemente.

    ✓ Resposta:
    import asyncio
import aiohttp

async def fetch_status(session, url):
    async with session.get(url) as resp:
        return url, resp.status

async def main():
    urls = [
        "https://example.com",
        "https://httpbin.org/get",
        "https://jsonplaceholder.typicode.com/todos/1"
    ]
    async with aiohttp.ClientSession() as session:
        tarefas = [fetch_status(session, url) for url in urls]
        resultados = await asyncio.gather(*tarefas)
        for url, status in resultados:
            print(f"{url}: {status}")

asyncio.run(main())

  5. Crie uma corrotina que tente executar uma operação com asyncio.timeout (Python 3.11+) e trate o TimeoutError. Use um atraso de 2 segundos com timeout de 1 segundo.

    ✓ Resposta:
    import asyncio

async def main():
    try:
        async with asyncio.timeout(1):
            await asyncio.sleep(2)
    except TimeoutError:
        print("Operação cancelada por timeout!")

asyncio.run(main())