A programação assíncrona revolucionou a forma como desenvolvemos aplicações em Python. Se você precisa fazer múltiplas requisições HTTP, processar arquivos em paralelo ou construir APIs de alta performance, o async/await é a ferramenta essencial que você precisa dominar.

Neste guia completo, você vai aprender desde os conceitos fundamentais até técnicas avançadas de programação assíncrona, com exemplos práticos que você pode aplicar imediatamente nos seus projetos.

O Que é Programação Assíncrona?

Antes de mergulharmos no código, é fundamental entender o que torna a programação assíncrona tão poderosa. Em Python, temos dois modelos principais de execução:

Programação Síncrona (tradicional): Cada operação espera a anterior terminar antes de começar. É como fazer uma fila no banco — você só é atendido quando a pessoa da frente termina.

Programação Assíncrona: Permite que várias operações aconteçam simultaneamente. É como vários caixas trabalhando ao mesmo tempo, atendendo várias pessoas em paralelo.

A diferença de performance pode ser dramático. Enquanto um código síncrono que faz 100 requisições HTTP pode levar 50 segundos (100 × 0.5s), a versão assíncrona pode completar tudo em menos de 1 segundo, executando as requisições em paralelo.

Introduzindo o asyncio

O módulo asyncio é o coração da programação assíncrona em Python. Introduzido na versão 3.4 e significativamente melhorado na 3.5 com a sintaxe native async/await, ele fornece a infraestrutura necessária para criar aplicações concorrentes eficientes.

Instalação e Verificação

O asyncio já vem instalado com Python 3.4+. Não precisa de pip install. Para verificar a versão disponível:

import asyncio

print(asyncio.__version__)

Primeiro Exemplo: "Hello Async"

import asyncio

async def hello():
    print("Olá!")
    await asyncio.sleep(1)  # Simula operação assíncrona
    print("Mundo!")

# Executar a corrotina
asyncio.run(hello())

Note a diferença: usamos async def para definir uma corrotina e await para chamar operações assíncronas. O asyncio.run() é o ponto de entrada que executa o loop de eventos.

Entendendo Corrotinas (Coroutines)

As corrotinas são a base da programação assíncrona em Python. Uma corrotina é uma função especial que pode pausar sua execução e retomar depois, permitindo que outras tarefas rodem enquanto ela está "esperando".

Definindo Corrotinas

import asyncio

async def buscar_dados():
    print("Iniciando busca...")
    await asyncio.sleep(2)  # Simula API call
    return {"dados": "importantes"}

async def processar():
    print("Processando...")
    await asyncio.sleep(1)
    return "concluído"

async def main():
    # Executar corrotinas
    resultado = await buscar_dados()
    print(f"Resultado: {resultado}")

    proc = await processar()
    print(f"Status: {proc}")

asyncio.run(main())

A Diferença Entre async def e def

A principal diferença entre funções síncronas e assíncronas:

# Função síncrona tradicional
def funcao_normal():
    return "valor"

# Corrotina (função assíncrona)
async def funcao_assincrona():
    return "valor"

# Você NÃO pode usar await em funções normais
# E NÃO pode usar funções normais onde await é esperado

Executando Tarefas em Paralelo

Um dos maiores benefícios da programação assíncrona é a capacidade de executar múltiplas tarefas simultaneamente. O asyncio fornece várias formas de fazer isso.

Usando asyncio.gather()

O asyncio.gather() permite executar múltiplas corrotinas concorrentemente e esperar que todas terminem:

import asyncio
import time

async def tarefa_demorada(nome, duracao):
    print(f"Iniciando {nome}")
    await asyncio.sleep(duracao)
    print(f"{nome} concluída!")
    return f"{nome} finalizada"

async def main():
    inicio = time.time()

    # Executar 3 tarefas em paralelo
    resultados = await asyncio.gather(
        tarefa_demorada("Tarefa A", 2),
        tarefa_demorada("Tarefa B", 3),
        tarefa_demorada("Tarefa C", 1),
    )

    tempo_total = time.time() - inicio

    print(f"Tempo total: {tempo_total:.2f}s")
    print(f"Resultados: {resultados}")

asyncio.run(main())

Resultado impressionante: Embora cada tarefa leve 2+3+1=6 segundos individualmente, o tempo total foi de apenas ~3 segundos porque elas rodaram em paralelo!

Usando asyncio.create_task()

Para executar tarefas em background sem esperar imediatamente:

import asyncio

async def tarefa_bg(nome):
    await asyncio.sleep(2)
    return f"{nome} pronta"

async def main():
    # Criar tarefa sem bloquear
    tarefa1 = asyncio.create_task(tarefa_bg("Job 1"))
    tarefa2 = asyncio.create_task(tarefa_bg("Job 2"))

    print("Tarefas criadas, continuando...")

    # Fazendo outras coisas enquanto as tarefas rodam
    await asyncio.sleep(0.5)
    print("Fizemos outra coisa!")

    # Agora sim, esperar pelos resultados
    resultado1 = await tarefa1
    resultado2 = await tarefa2

    print(resultado1, resultado2)

asyncio.run(main())

Timeouts e Tratamento de Erros

Em aplicações assíncronas, é crucial lidar com operações que podem demorar muito ou falhar. O asyncio oferece ferramentas poderosas para isso.

Definindo Timeout

import asyncio

async def operacao_lenta():
    await asyncio.sleep(10)
    return "Sucesso!"

async def main():
    try:
        #超时设置为3秒
        resultado = await asyncio.wait_for(operacao_lenta(), timeout=3)
        print(resultado)
    except asyncio.TimeoutError:
        print("Operação expirou!")

asyncio.run(main())

Tratamento de Exceções

import asyncio

async def operacao_falivel():
    await asyncio.sleep(1)
    raise ValueError("Algo deu errado!")

async def main():
    try:
        await operacao_falivel()
    except ValueError as e:
        print(f"Erro capturado: {e}")
    finally:
        print("Cleanup executado")

asyncio.run(main())

Requisições HTTP Assíncronas

Um dos casos de uso mais comuns de async/await é fazer múltiplas requisições HTTP. Para isso, usamos bibliotecas como aiohttp ou httpx.

Instalando httpx

pip install httpx

Exemplo Prático: Buscar Dados de Multiple APIs

import asyncio
import httpx
import time

async def buscar_usuario(client, user_id):
    """Buscar um usuário específico"""
    response = await client.get(f"https://jsonplaceholder.typicode.com/users/{user_id}")
    return response.json()

async def buscar_posts_usuario(client, user_id):
    """Buscar posts de um usuário"""
    response = await client.get(f"https://jsonplaceholder.typicode.com/posts?userId={user_id}")
    return response.json()

async def main():
    async with httpx.AsyncClient() as client:
        inicio = time.time()

        # Buscar dados de 5 usuários em paralelo
        tarefas = []
        for i in range(1, 6):
            usuario = buscar_usuario(client, i)
            posts = buscar_posts_usuario(client, i)
            tarefas.append(usuario)
            tarefas.append(posts)

        resultados = await asyncio.gather(*tarefas)

        tempo = time.time() - inicio

        print(f"Buscou {len(resultados)} dados em {tempo:.2f}s")
        print(f"Primeiro usuário: {resultados[0]['name']}")

asyncio.run(main())

Este código busca dados de 5 usuários e seus posts — um total de 10 requisições — em poucos milissegundos, muito mais rápido do que fazer cada request sequencialmente.

Semáforos para Limitar Concorrência

Às vezes você quer permitir concorrência, mas com um limite. Isso é útil para não sobrecarregar uma API ou servidor:

import asyncio

async def baixar_arquivo(semaphore, numero):
    async with semaphore:
        print(f"Baixando arquivo {numero}...")
        await asyncio.sleep(1)  # Simula download
        return f"Arquivo {numero} baixado"

async def main():
    # Limitar a 3 downloads simultâneos
    semaphore = asyncio.Semaphore(3)

    tarefas = [baixar_arquivo(semaphore, i) for i in range(10)]
    resultados = await asyncio.gather(*tarefas)

    for r in resultados:
        print(r)

asyncio.run(main())

Filas Assíncronas (Async Queue)

Para cenários de producer-consumer ou tarefas em background:

import asyncio
import random

async def produtor(queue, itens):
    for item in itens:
        await asyncio.sleep(random.random())
        await queue.put(item)
        print(f"Produzido: {item}")
    await queue.put(None)  # Sinal de fim

async def consumidor(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumindo: {item}")
        await asyncio.sleep(0.5)
        queue.task_done()

async def main():
    queue = asyncio.Queue()

    await asyncio.gather(
        produtor(queue, ["A", "B", "C", "D", "E"]),
        consumidor(queue)
    )

asyncio.run(main())

Event Loops Avançados

O event loop é o coração do asyncio. Entender como ele funciona permite otimizar suas aplicações.

Executando Múltiplos Loops

import asyncio

async def tarefa1():
    await asyncio.sleep(1)
    return "Tarefa 1"

async def tarefa2():
    await asyncio.sleep(1)
    return "Tarefa 2"

async def main():
    #gather executa corrotinas no mesmo loop
    resultados = await asyncio.gather(tarefa1(), tarefa2())
    print(resultados)

# Loop padrão (recomendado para Python 3.10+)
asyncio.run(main())

Loop Personalizado

import asyncio

async def tarefa(duracao, nome):
    print(f"{nome} iniciando")
    await asyncio.sleep(duracao)
    print(f"{nome} terminada")
    return nome

async def main():
    # Criar loop customizado
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    try:
        resultado = await loop.run_until_complete(
            asyncio.gather(
                tarefa(1, "A"),
                tarefa(2, "B"),
            )
        )
        print(f"Resultados: {resultado}")
    finally:
        loop.close()

main()

Async Context Managers

Similar aos context managers síncronos (with), mas para operações assíncronas:

import asyncio

class ConectorAsync:
    async def __aenter__(self):
        print("Conectando...")
        await asyncio.sleep(0.5)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Desconectando...")
        await asyncio.sleep(0.2)

    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Resultado: {sql}"

async def main():
    async with ConectorAsync() as conn:
        resultado = await conn.query("SELECT * FROM usuarios")
        print(resultado)

asyncio.run(main())

Melhores Práticas

Agora que você conhece os fundamentos, aqui estão algumas práticas essenciais:

1. Use async/await Sempre Que Possível

# Bom: usar bibliotecas assíncronas
import aiofiles
import aiohttp

async def ler_arquivo():
    async with aiofiles.open('arquivo.txt', 'r') as f:
        return await f.read()

2. Evite Bloquear o Event Loop

# Ruim: usar código síncrono bloqueante
import time

async def problema():
    time.sleep(10)  # BLOQUEIA o event loop!

# Bom: usar versões assíncronas
async def solucao():
    await asyncio.sleep(10)  # Não bloqueia

3. Use Contexto de Sessão

import httpx

# Sempre use async with para clients HTTP
async with httpx.AsyncClient() as client:
    response = await client.get(url)

4. Defina Timeouts

import asyncio

await asyncio.wait_for(operacao(), timeout=5.0)

Aplicações Reais

A programação assíncrona é usada em diversos cenários:

  • APIs de alta performance: FastAPI e Sanic usam async nativamente para Handle milhares de requisições simultâneas
  • Web scraping: Gather permite coletar dados de múltiplas páginas simultaneamente
  • Chatbots e messengers: Responder vários usuários sem bloquear
  • IoT e streaming: Processar dados de múltiplos sensores em tempo real
  • Microservices: Comunicação eficiente entre serviços

Async vs Threads vs Multiprocessing

Entenda quando usar cada abordagem:

Async I/O: Ideal para operações de I/O intensivo (rede, arquivo, banco). Baixo overhead, excelenty para milhares de conexões simultâneas.

Threads: Útil para operações CPU-bound com necessidade de compartilhar memória. Python tem o GIL como limitação para CPU puro.

Multiprocessing: Melhor para CPU-bound heavy processing (machine learning, processamentoy de vídeo). Cada processo tem sua própria memória.

Próximos Passos

Agora que você dominou async/await, continue aprendendo:

A programação assíncrona é uma habilidade essential para qualquer desenvolvedor Python moderno. Comece a implementar em seus projetos hoje mesmo e sinta a diferença de performance!

Para mais conteúdo sobre Python e desenvolvimento web, continue acompanhando o Universo Python!