Nesta aula, vamos mergulhar no módulo asyncio do Python, essencial para programação assíncrona e concorrência. O asyncio permite escrever código que executa múltiplas tarefas de I/O de forma eficiente, usando uma única thread e um loop de eventos. Vamos explorar conceitos práticos como criação de tarefas, agregação de corrotinas, controle de tempo e operações de I/O assíncrono, com exemplos que você pode executar e adaptar.

Dominar asyncio é fundamental para desenvolver aplicações web de alta performance, clientes de rede, ferramentas de automação e qualquer sistema que precise lidar com muitas operações de entrada/saída simultaneamente. Ao final desta aula, você terá uma base sólida para usar asyncio em seus projetos.

Tarefas (create_task)

Em asyncio, uma corrotina é uma função definida com async def que pode ser pausada e retomada. Para executar uma corrotina de forma concorrente com outras, usamos asyncio.create_task(), que agenda a execução da corrotina no loop de eventos e retorna um objeto Task. Isso permite que várias corrotinas sejam executadas de forma intercalada, sem bloquear o loop.

Vejamos um exemplo simples:

import asyncio

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

async def main():
    # Cria duas tarefas que rodam concorrentemente
    tarefa1 = asyncio.create_task(dizer_oi("Alice", 2))
    tarefa2 = asyncio.create_task(dizer_oi("Bob", 1))

    # Aguarda ambas terminarem
    await tarefa1
    await tarefa2

asyncio.run(main())

Ao executar, a mensagem de Bob aparece primeiro (após 1 segundo) e a de Alice depois (após 2 segundos), mostrando a execução concorrente. Note que usamos await tarefa para aguardar a conclusão. Se não aguardarmos, a tarefa pode ser cancelada quando o programa terminar.

Uma prática comum é aguardar várias tarefas com asyncio.gather(), que veremos a seguir. Também é importante lembrar que create_task só funciona dentro de uma corrotina em execução (ou seja, dentro de async).

gather

asyncio.gather() é uma função que agrega várias corrotinas ou tarefas e as executa de forma concorrente, retornando uma lista com os resultados na mesma ordem em que foram passadas. É ideal quando você precisa executar várias operações assíncronas e coletar seus resultados.

Exemplo:

import asyncio

async def calcular_quadrado(n):
    await asyncio.sleep(1)  # simula operação demorada
    return n * n

async def main():
    resultados = await asyncio.gather(
        calcular_quadrado(2),
        calcular_quadrado(3),
        calcular_quadrado(4),
    )
    print(resultados)  # [4, 9, 16]

asyncio.run(main())

Se uma das corrotinas lançar uma exceção, gather propaga a primeira exceção encontrada, cancelando as demais (por padrão). Para evitar isso, use return_exceptions=True, que retorna as exceções como objetos na lista de resultados.

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

async def main():
    resultados = await asyncio.gather(
        falhar(),
        calcular_quadrado(2),
        return_exceptions=True,
    )
    print(resultados)  # [ValueError('Erro!'), 4]

Isso é útil para lidar com falhas parciais sem interromper todas as tarefas.

Timeouts

Em muitas aplicações, é necessário limitar o tempo de execução de uma operação assíncrona. O asyncio oferece asyncio.timeout() (Python 3.11+) e asyncio.wait_for() para esse fim. wait_for aguarda uma corrotina ou tarefa e lança asyncio.TimeoutError se o tempo limite for excedido.

Exemplo com wait_for:

import asyncio

async def operacao_demorada():
    await asyncio.sleep(10)
    return "Concluído"

async def main():
    try:
        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())

Já em Python 3.11+, a forma recomendada é usar asyncio.timeout() como um gerenciador de contexto assíncrono:

import asyncio

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

Note que asyncio.timeout() lança TimeoutError (não asyncio.TimeoutError). O uso de timeout é mais limpo e integra-se bem com outras funcionalidades, como cancelamento.

É importante tratar os timeouts adequadamente para não deixar recursos pendentes ou tarefas órfãs.

I/O assíncrono

O asyncio é especialmente útil para operações de I/O, como leitura/escrita de arquivos, comunicação de rede e chamadas HTTP. Vamos ver um exemplo de leitura de arquivos de forma assíncrona usando a biblioteca aiofiles (que precisa ser instalada: pip install aiofiles).

import asyncio
import aiofiles

async def ler_arquivo(caminho):
    async with aiofiles.open(caminho, mode='r') as f:
        conteudo = await f.read()
    return conteudo

async def main():
    conteudo = await ler_arquivo('exemplo.txt')
    print(conteudo[:100])  # imprime os primeiros 100 caracteres

asyncio.run(main())

Da mesma forma, para requisições HTTP assíncronas, usamos bibliotecas como aiohttp ou httpx. Exemplo com httpx (instale com pip install httpx):

import asyncio
import httpx

async def buscar_url(url):
    async with httpx.AsyncClient() as cliente:
        resposta = await cliente.get(url)
        return resposta.status_code

async def main():
    urls = ['https://example.com', 'https://httpbin.org/get']
    statuses = await asyncio.gather(*(buscar_url(url) for url in urls))
    print(statuses)

asyncio.run(main())

Essas operações não bloqueiam o loop de eventos, permitindo que outras tarefas sejam executadas enquanto a operação de I/O está pendente. Isso é fundamental para criar aplicações escaláveis.

Boas práticas e observações finais

Ao trabalhar com asyncio, tenha em mente:

  • Sempre use asyncio.run() como ponto de entrada, exceto em contextos onde o loop já está rodando (como em frameworks web).
  • Evite misturar código síncrono que bloqueia (como time.sleep) dentro de corrotinas; prefira asyncio.sleep.
  • Para tarefas de longa duração, considere usar asyncio.create_task e aguardar com gather.
  • Use asyncio.timeout para controlar operações que podem travar.
  • Em aplicações reais, use bibliotecas assíncronas para I/O (aiofiles, aiohttp, asyncpg, etc.) para aproveitar todo o potencial do asyncio.

Exercícios

  1. Escreva uma corrotina contar(nome, atraso) que imprime o nome, espera atraso segundos e imprime o nome novamente. Use create_task para executar duas dessas corrotinas com atrasos diferentes em main e aguarde ambas.

    ✓ Resposta: 
    import asyncio

async def contar(nome, atraso):
    print(f"{nome} iniciou")
    await asyncio.sleep(atraso)
    print(f"{nome} terminou")

async def main():
    t1 = asyncio.create_task(contar("A", 2))
    t2 = asyncio.create_task(contar("B", 1))
    await t1
    await t2

asyncio.run(main())

  2. Crie uma função assíncrona soma(a, b) que retorna a soma após 1 segundo. Use asyncio.gather para somar três pares de números e imprimir os resultados.

    ✓ Resposta: 
    import asyncio

async def soma(a, b):
    await asyncio.sleep(1)
    return a + b

async def main():
    resultados = await asyncio.gather(
        soma(1, 2),
        soma(3, 4),
        soma(5, 6),
    )
    print(resultados)  # [3, 7, 11]

asyncio.run(main())

  3. Usando asyncio.wait_for, crie uma corrotina que dorme 5 segundos e um main que espera 2 segundos. Capture a exceção e imprima uma mensagem.

    ✓ Resposta: 
    import asyncio

async def dormir():
    await asyncio.sleep(5)
    return "acordou"

async def main():
    try:
        resultado = await asyncio.wait_for(dormir(), timeout=2)
        print(resultado)
    except asyncio.TimeoutError:
        print("A operação demorou demais!")

asyncio.run(main())

  4. Escreva um programa que use aiohttp para fazer requisições GET a duas URLs diferentes simultaneamente e imprima o status de cada uma. (Assuma que aiohttp está instalado.)

    ✓ Resposta: 
    import asyncio
import aiohttp

async def buscar_status(url):
    async with aiohttp.ClientSession() as sessao:
        async with sessao.get(url) as resp:
            return resp.status

async def main():
    urls = ['https://example.com', 'https://httpbin.org/get']
    statuses = await asyncio.gather(*(buscar_status(url) for url in urls))
    print(statuses)

asyncio.run(main())

  5. Explique a diferença entre asyncio.create_task e asyncio.gather e quando usar cada um.

    ✓ Resposta:

    create_task agenda uma única corrotina para execução concorrente, retornando um objeto Task que pode ser aguardado individualmente. gather recebe várias corrotinas/tarefas, executa-as concorrentemente e retorna uma lista de resultados. Use create_task quando você precisa de controle individual sobre cada tarefa (por exemplo, cancelar uma delas) ou quando quer aguardar tarefas em momentos diferentes. Use gather quando você quer executar um conjunto de corrotinas e aguardar todas juntas, coletando os resultados.

Referências