Recentemente eu recebi a tarefa de fazer 300.000 requisições em uma API. A primeira ideia foi usar a biblioteca requests, mas logo duvidei: com esse volume, o processo levaria tempo demais para terminar.
Foi aí que pensei em assíncronismo, utilizando as bibliotecas aiohttp e asyncio. Em vez de enviar uma requisição, esperar a resposta e só então enviar a próxima, podemos disparar centenas ou milhares de requisições por segundo, aproveitando ao máximo a capacidade de execução concorrente do Python.
Se você também precisa lidar com grandes volumes de requisições HTTP, talvez esteja preso no trânsito da programação síncrona. Neste exemplo, buscando cotações do dólar com Python através de uma API, vamos ver na prática as diferenças entre a programação síncrona e assíncrona, e como o assíncronismo pode multiplicar sua eficiência.
Requisições HTTP: o Cenário, API de Cotação e o Objetivo
Nosso objetivo é buscar a cotação de fechamento (USD-BRL) dos últimos 90 dias. Para simular um cenário de “requisições em massa”, faremos um loop e pediremos um dia de cada vez.
Usaremos a BrasilAPI por sua simplicidade. O endpoint para um dia específico é:
https://brasilapi.com.br/api/cambio/v1/cotacao/USD/2025-10-22Vamos precisar das seguintes bibliotecas:
requests: Para testar o método síncrono tradicional.aiohttpeasyncio: Para o método assíncrono.pandas: Para organizar os dados e salvar em CSV facilmente.datetime: Para gerar a lista de datas.
Requests: requisições síncronas no Python
Este é o método “padrão” que a maioria dos iniciantes aprende. Usamos a biblioteca requests dentro de um loop for.
Como funciona:
- Começa o loop (Dia 1).
- Faz a requisição para o Dia 1.
- ESPERA a API responder.
- Processa a resposta do Dia 1.
- Começa o loop (Dia 2).
- Faz a requisição para o Dia 2.
- ESPERA… (e assim por diante).
De certa forma, o script passa a maior parte do tempo “de braços cruzados”, apenas esperando o retorno da API. Na correria do dia a dia isso é longe de ser o ideal, embora consiga resolver alguns cenários de menor demanda.
Script de exemplo: requisições síncronas
No exemplo abaixo, vamos utilizar a BrasilAPI para buscar a cotação de fechamento do dólar nos últimos 90 dias e armazenar os resultados em um arquivo .csv.
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
def get_dates_range(days):
"""Gera uma lista de datas (YYYY-MM-DD) dos últimos 'days' dias."""
today = datetime.now()
dates = []
for i in range(days):
day = today - timedelta(days=i)
dates.append(day.strftime('%Y-%m-%d'))
return dates
def find_closing_quote(cotacoes_list):
"""
Encontra a cotação de fechamento PTAX.
Se não encontrar, retorna a última cotação do dia.
"""
# Tenta encontrar o fechamento PTAX (busca de trás para frente)
for c in reversed(cotacoes_list):
if c.get('tipo_boletim') == 'FECHAMENTO PTAX':
return float(c['cotacao_venda'])
# Plano B: Se não achar, pega a 'cotacao_venda' do último boletim
if cotacoes_list:
return float(cotacoes_list[-1]['cotacao_venda'])
return None
def fetch_sync(dates):
"""Busca as cotações de forma síncrona."""
print("Iniciando busca SÍNCRONA (BrasilAPI)...")
api_base_url = "https://brasilapi.com.br/api/cambio/v1/cotacao/USD/"
results = []
for date in dates:
url = f"{api_base_url}{date}"
try:
response = requests.get(url, timeout=10)
# Verifica se a requisição foi bem-sucedida
if response.status_code == 200:
data = response.json()
if data.get('cotacoes'):
cotacao_fechamento = find_closing_quote(data['cotacoes'])
if cotacao_fechamento:
results.append({
'data_buscada': date,
'cotacao_fechamento': cotacao_fechamento
})
else:
print(f"Nenhum boletim válido encontrado para {date}.")
else:
print(f"Resposta OK, mas sem cotações para {date}.")
# A API retorna 404 para fins de semana ou feriados
elif response.status_code == 404:
print(f"Sem cotação para {date} (Fim de semana/feriado).")
else:
print(f"Erro {response.status_code} ao buscar data {date}.")
except requests.exceptions.RequestException as e:
print(f"Erro de conexão ao buscar data {date}: {e}")
return results
# --- Execução Principal ---
if __name__ == "__main__":
NUM_DIAS = 90 # Buscando 90 dias
datas = get_dates_range(NUM_DIAS)
start_time = time.perf_counter()
cotacoes = fetch_sync(datas)
end_time = time.perf_counter()
# Salvando em CSV
if cotacoes:
df = pd.DataFrame(cotacoes)
df.to_csv('cotacoes_sync_brasilapi.csv', index=False)
print(f"\nTotal de {len(cotacoes)} cotações salvas em 'cotacoes_sync_brasilapi.csv'.")
else:
print("\nNenhuma cotação foi baixada.")
print("-" * 30)
print(f"Busca SÍNCRONA concluída.")
print(f"Tempo total: {end_time - start_time:.2f} segundos.")Hoje não estamos utilizando o Google Colab, mas sim um ambiente Python no Visual Studio Code. Esse foi o tempo de execução do script síncrono:
104 segundos para efetuar 90 cotações.
aiohttp e asyncio: requisições assíncronas
Aqui, usamos asyncio (como gerenciador de eventos) e aiohttp (o requests assíncrono).
Como funciona:
- O
asynciocria um “loop de eventos”. - Dizemos a ele: “aqui está uma lista de 90 tarefas (requisições)”.
- O
asyncioinicia TODAS as 90 requisições quase simultaneamente. - O script então ESPERA pela primeira que responder.
- Enquanto espera pelas outras, ele processa a que chegou.
- O tempo total é ditado pela requisição mais lenta, não pela soma de todas.
A limitação aqui é computacional (do seu lado): ou seja, se você tem muitas tarefas complexas, o seu hardware precisa suportar. E aqui no nosso caso, também da API que está sendo chamada: muitos sistemas impõe limites de requisições por minuto, para evitar sobrecarga. Se atente a isso para evitar bloqueios.
Exemplo para requisições assíncronas
Para efetuar requisições de forma assíncrona, usaremos o script abaixo:
import aiohttp
import asyncio
import pandas as pd
import time
from datetime import datetime, timedelta
def get_dates_range(days):
"""Gera uma lista de datas (YYYY-MM-DD) dos últimos 'days' dias."""
today = datetime.now()
dates = []
for i in range(days):
day = today - timedelta(days=i)
dates.append(day.strftime('%Y-%m-%d'))
return dates
def find_closing_quote(cotacoes_list):
"""
Encontra a cotação de fechamento PTAX.
Se não encontrar, retorna a última cotação do dia.
"""
for c in reversed(cotacoes_list):
if c.get('tipo_boletim') == 'FECHAMENTO PTAX':
return float(c['cotacao_venda'])
if cotacoes_list:
return float(cotacoes_list[-1]['cotacao_venda'])
return None
async def fetch_one_quote(session, date):
"""Busca uma única cotação de forma assíncrona."""
api_base_url = "https://brasilapi.com.br/api/cambio/v1/cotacao/USD/"
url = f"{api_base_url}{date}"
try:
async with session.get(url, timeout=10) as response:
if response.status == 200:
data = await response.json()
if data.get('cotacoes'):
cotacao_fechamento = find_closing_quote(data['cotacoes'])
if cotacao_fechamento:
return {
'data_buscada': date,
'cotacao_fechamento': cotacao_fechamento
}
else:
print(f"Nenhum boletim válido encontrado para {date}.")
else:
print(f"Resposta OK, mas sem cotações para {date}.")
elif response.status == 404:
print(f"Sem cotação para {date} (Fim de semana/feriado).")
else:
print(f"Erro {response.status} ao buscar data {date}.")
except asyncio.TimeoutError:
print(f"Timeout ao buscar data {date}.")
except aiohttp.ClientError as e:
print(f"Erro de conexão ao buscar data {date}: {e}")
return None
async def fetch_async(dates):
"""Cria tarefas para todas as datas e as executa concorrentemente."""
print("Iniciando busca ASSÍNCRONA (BrasilAPI)...")
tasks = []
async with aiohttp.ClientSession() as session:
for date in dates:
tasks.append(fetch_one_quote(session, date))
results = await asyncio.gather(*tasks)
return [r for r in results if r] # Filtra resultados nulos (None)
# --- Execução Principal ---
if __name__ == "__main__":
NUM_DIAS = 90
datas = get_dates_range(NUM_DIAS)
start_time = time.perf_counter()
# Use asyncio.run() para gerenciar o loop automaticamente
cotacoes = asyncio.run(fetch_async(datas))
end_time = time.perf_counter()
# Salvando em CSV
if cotacoes:
df = pd.DataFrame(cotacoes)
df.to_csv('cotacoes_async_brasilapi.csv', index=False)
print(f"\nTotal de {len(cotacoes)} cotações salvas em 'cotacoes_async_brasilapi.csv'.")
else:
print("\nNenhuma cotação foi baixada.")
print("-" * 30)
print(f"Busca ASSÍNCRONA concluída.")
print(f"Tempo total: {end_time - start_time:.2f} segundos.")
Agora temos 90 requisições concluídas em 10.5 segundos. A execução ficou ~10x mais rápida.
Se tomarmos o exemplo inicial das 300.000 requisições, antes levaria 96h e agora passaria para 9,7h – o que ainda é muito.
| Métrica | Antes | Depois | Melhora |
|---|---|---|---|
| Tempo por 90 req | 104 s | 10,5 s | 9,9x |
| Tempo por requisição | 1,155 s | 0,1167 s | −89,9% |
| Tempo p/ 300.000 req | 96 h | 9,7 h | -86 h |
Conclusão: Quando usar cada um?
A programação assíncrona não é “bala de prata” e adiciona uma leve complexidade ao código (o uso de async/await). A chave é entender onde está o gargalo.
Para todos os loops/for que você tiver em Python, indicamos o uso do tqdm para entender se você tem um gargalo, basta olhar para sua taxa de iteração por segundo.
Regra de Ouro:
- Use Síncrono (
requests) quando sua tarefa é limitada pela CPU (cálculos pesados, processamento de dados local) ou para scripts muito simples com poucas chamadas de rede.- Use Assíncrono (
aiohttp) quando sua tarefa é limitada por I/O (Input/Output). Isso inclui esperar por rede (APIs, web scraping), ler arquivos grandes, ou acessar bancos de dados.
Para extração de dados em massa, a programação assíncrona não é apenas uma opção, é uma necessidade para criar scripts eficientes e rápidos. Indicamos também que leia sobre os conceitos de concorrência x paralelismo.
