Python asyncio: async e await

O módulo asyncio do Python permite escrever código concorrente em uma única thread usando as palavras-chave async e await. Em vez de bloquear enquanto aguarda respostas de rede ou leituras de arquivos, um programa asyncio suspende a task em espera e imediatamente muda para outro trabalho — retomando quando o resultado estiver pronto. Isso o torna a ferramenta certa para programas limitados por I/O, como raspadores web, clientes de API e servidores de chat.

Este capítulo cobre:

• O que são funções async (corrotinas) e como elas diferem de funções regulares
• O event loop e como o asyncio agenda o trabalho
• Aguardar resultados, executar tasks concorrentemente com asyncio.gather e asyncio.create_task
• Tratar exceções e timeouts dentro de código async
• O asyncio.Queue para padrões produtor-consumidor
• Quando usar asyncio e quando recorrer a threading

Por que o asyncio existe

Considere um programa que chama duas APIs uma após a outra:

import time

def fetch(name, delay):
    time.sleep(delay)          # blocks the whole program
    return f'data from {name}'

start = time.perf_counter()
r1 = fetch('API A', 1)
r2 = fetch('API B', 1)
print(f'Done in {time.perf_counter() - start:.1f}s')
# Done in 2.0s

Ambas as chamadas são executadas sequencialmente — 2 segundos no total, mesmo que cada chamada precise apenas de 1 segundo de espera. Com o asyncio, o programa pausa fetch('API A', ...) enquanto aguarda, inicia fetch('API B', ...) imediatamente, e ambas terminam em cerca de 1 segundo:

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)   # suspends only this coroutine
    return f'data from {name}'

async def main():
    start = time.perf_counter()
    r1, r2 = await asyncio.gather(fetch('API A', 1), fetch('API B', 1))
    print(f'Done in {time.perf_counter() - start:.1f}s')
    # Done in 1.0s

asyncio.run(main())

Corrotinas: async def e await

Uma função definida com async def é chamada de função corrotina. Chamá-la não executa o corpo imediatamente — ela retorna um objeto corrotina que deve ser conduzido pelo event loop.

async def greet(name):
    print(f'Hello, {name}!')

# Calling it returns a coroutine object, nothing is printed yet
coro = greet('World')
print(type(coro))   # <class 'coroutine'>

# Run it properly
import asyncio
asyncio.run(greet('World'))
# Hello, World!

Dentro de uma corrotina, await suspende a execução até que o aguardável (outra corrotina, uma Task ou um Future) produza um resultado. O event loop fica livre para executar outras corrotinas enquanto uma está suspensa.

import asyncio

async def step_one():
    print('Step 1: start')
    await asyncio.sleep(1)     # suspend for 1 second
    print('Step 1: end')
    return 'result-1'

async def main():
    value = await step_one()   # wait for step_one to finish
    print(value)

asyncio.run(main())
# Step 1: start
# Step 1: end
# result-1

O que você pode aguardar

• Outra corrotina async def
• Um asyncio.Task (criado com asyncio.create_task)
• Um asyncio.Future
• Qualquer object com um método __await__

Você não pode usar await fora de uma função async def.

O event loop

O event loop é o agendador do asyncio. Ele mantém uma fila de corrotinas e tasks, executa cada uma até atingir um await, e então muda para o próximo item pronto. Normalmente há um event loop por thread.

asyncio.run(coro) é o ponto de entrada padrão para programas asyncio. Ele cria um novo event loop, executa a corrotina fornecida até a conclusão, fecha o loop e retorna o resultado:

import asyncio

async def compute():
    await asyncio.sleep(0)   # yield control once
    return 6 * 7

result = asyncio.run(compute())
print(result)   # 42

Para a maioria das aplicações, você nunca precisa gerenciar o loop diretamente — asyncio.run cuida da criação e do encerramento.

Executando tasks concorrentemente

asyncio.gather

asyncio.gather(*coroutines) agenda todas as corrotinas fornecidas para serem executadas concorrentemente e retorna seus resultados na mesma ordem:

import asyncio

async def fetch_data(name, delay):
    print(f'Start fetching {name}')
    await asyncio.sleep(delay)
    print(f'Done fetching {name}')
    return f'data from {name}'

async def main():
    results = await asyncio.gather(
        fetch_data('API A', 1),
        fetch_data('API B', 2),
        fetch_data('API C', 1),
    )
    print(results)

asyncio.run(main())
# Start fetching API A
# Start fetching API B
# Start fetching API C
# Done fetching API A
# Done fetching API C
# Done fetching API B
# ['data from API A', 'data from API B', 'data from API C']

Todas as três corrotinas iniciam imediatamente. O tempo total decorrido corresponde à corrotina mais lenta (2 s), não à soma (4 s).

asyncio.create_task

asyncio.create_task(coro) envolve uma corrotina em uma Task e a agenda para execução em breve. Diferentemente do gather, criar uma task a dispara em segundo plano enquanto a corrotina atual continua sendo executada:

import asyncio

async def background_job(name, delay):
    print(f'{name}: start')
    await asyncio.sleep(delay)
    print(f'{name}: end')
    return f'{name} done'

async def main():
    t1 = asyncio.create_task(background_job('Task A', 1))
    t2 = asyncio.create_task(background_job('Task B', 2))

    # Both tasks are already scheduled; await collects their results
    result1 = await t1
    result2 = await t2
    print(result1, result2)

asyncio.run(main())
# Task A: start
# Task B: start
# Task A: end
# Task B: end
# Task A done Task B done

Use create_task quando quiser que uma task inicie imediatamente e você planeja coletar seu resultado (ou cancelá-la) mais tarde. Use gather quando quiser lançar um grupo fixo de corrotinas e aguardar todas juntas.

Saída intercalada

Uma maneira útil de ver o event loop em ação é observar como as tasks se intercalam:

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f'{name}: {i}')
        await asyncio.sleep(1)
    print(f'{name}: done!')

async def main():
    await asyncio.gather(
        count_down('Task A', 3),
        count_down('Task B', 2),
    )

asyncio.run(main())
# Task A: 3
# Task B: 2
# Task A: 2
# Task B: 1
# Task A: 1
# Task B: done!
# Task A: done!

Ambas as tasks compartilham uma thread; o event loop alterna entre elas a cada await asyncio.sleep(1).

Tratando exceções

Exceções levantadas dentro de uma corrotina se propagam por meio de await assim como no código síncrono. Use um bloco try/except normal:

import asyncio

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError('something went wrong')

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f'Caught: {e}')

asyncio.run(main())
# Caught: something went wrong

Ao usar asyncio.gather, se uma corrotina levantar uma exceção, as outras não são canceladas por padrão, mas a exceção é relançada quando você awaita a chamada gather. Passe return_exceptions=True para coletar exceções como valores de retorno:

import asyncio

async def good():
    return 'ok'

async def bad():
    raise RuntimeError('oops')

async def main():
    results = await asyncio.gather(good(), bad(), return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f'Error: {r}')
        else:
            print(f'Result: {r}')

asyncio.run(main())
# Result: ok
# Error: oops

Timeouts com asyncio.wait_for

asyncio.wait_for(coro, timeout) executa uma corrotina e a cancela se não terminar dentro do número de segundos especificado, lançando asyncio.TimeoutError:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return 42

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print('Timed out — operation cancelled')

asyncio.run(main())
# Timed out — operation cancelled

Isso é importante para código de rede em produção, onde um servidor travado bloquearia uma task indefinidamente.

asyncio.Queue para padrões produtor-consumidor

asyncio.Queue é uma fila thread-safe e consciente de async. É ideal para desacoplar produtores (código que gera trabalho) de consumidores (código que o processa):

import asyncio

async def producer(queue):
    for i in range(1, 4):
        print(f'Produced item {i}')
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)   # sentinel to signal consumers to stop

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed item {item}')

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
# Produced item 1
# Consumed item 1
# Produced item 2
# Consumed item 2
# Produced item 3
# Consumed item 3

Para múltiplos consumidores, use queue.task_done() e queue.join() para saber quando todos os itens foram processados.

asyncio vs threading

Tanto o asyncio quanto o módulo threading do Python permitem que o trabalho avance concorrentemente, mas de maneiras diferentes:

Use asyncio quando você controla o código de I/O e pode usar bibliotecas compatíveis com async (ex.: aiohttp, asyncpg). Use threading quando depender de bibliotecas bloqueantes de terceiros que não podem ser tornadas async.

Para verdadeiro paralelismo de CPU, recorra a multiprocessing ou concurrent.futures.ProcessPoolExecutor.

Armadilhas comuns

Esquecer o await: Chamar uma função async sem await retorna um object corrotina e não faz nada. O Python emite um RuntimeWarning: coroutine '...' was never awaited para ajudar a detectar isso.

async def main():
    asyncio.sleep(1)   # BUG: returns a coroutine, does not sleep
    await asyncio.sleep(1)   # correct

Bloquear o event loop: Executar código síncrono lento (um loop intenso, uma chamada de rede bloqueante, time.sleep) dentro de uma corrotina congela o event loop inteiro. Envolva chamadas bloqueantes com asyncio.to_thread (Python 3.9+) para executá-las em um pool de threads sem bloquear:

import asyncio
import time

def blocking_task():
    time.sleep(2)   # simulates a slow blocking operation
    return 'done'

async def main():
    result = await asyncio.to_thread(blocking_task)
    print(result)

asyncio.run(main())
# done

Usar asyncio.run dentro de um loop em execução: Notebooks Jupyter já executam um event loop. Use await coro diretamente nas células do notebook, ou instale nest_asyncio para permitir loops aninhados.

Resumo de referência rápida

Prática