O Uso Da Tecnologia

Configurando automações com API REST: o que funciona e o que quebra

A maioria das pessoas que tenta implementar o uso da tecnologia para automação empresarial tropeça na mesma coisa: não entende como lidar com rate limits e tokens de sessão expirados. Já vi integrações de CRM com ERPs que funcionavam perfeitamente durante testes e paravam num domingo de manhã porque ninguém configurou refresh token automático. O problema nunca é a lógica principal do fluxo. É sempre a parte que ninguém pensa até dar erro. Eu comecei a mexer com integrações por volta de 2015, quando ainda usava SOAP pra conectar sistemas legacy. Hoje o cenário mudou, mas os mesmos erros continuam acontecendo. Vou explicar como montar uma automação básica com Python e requisições HTTP, sem complicação desnecessária.

O uso da tecnologia na prática: autenticação OAuth 2.0

O primeiro passo é configurar a autenticação. A maioria dos serviços modernos usa OAuth 2.0, que é basicamente um protocolo onde você pede permissão ao usuário, recebe um token de acesso com tempo de validade curto e um token de atualização para renovar quando expira. Parece simples. Na prática, muita gente esquece de salvar o token de atualização ou não implementa o refresh automático. Eu já perdi uma integração inteira porque o token venceu e o script simplesmente parou de funcionar. Ninguém percebeu porque os dados que já tinham sido processados apareciam normais no relatório. Vamos ao código. Você vai precisar de duas bibliotecas: requests e python-dotenv para cuidar das credenciais. Primeiro, crie um arquivo .env com suas credenciais. Coloque o client_id, client_secret, scope e as URLs de authorization e token do serviço que você está integrando. Nunca commit esse arquivo no GitHub. Já vi isso acontecer e as credenciais ficarem expostas por meses até alguém perceber.

Aqui está a estrutura básica do request de autenticação:

import requests
from dotenv import load_dotenv
import os

load_dotenv()

CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
TOKEN_URL = os.getenv("TOKEN_URL")
SCOPE = os.getenv("SCOPE", "read write")

def get_access_token():
    response = requests.post(
        TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "scope": SCOPE
        }
    )
    response.raise_for_status()
    return response.json()

Isso retorna um JSON com access_token, expires_in e geralmente refresh_token. Guarde tudo. O expires_in normalmente vem em segundos, então 3600 significa uma hora. A maioria dos serviços dá entre 15 minutos e 1 hora de validade pro token de acesso. Agora o ponto que todo mundo erra: o refresh. Você precisa verificar se o token vai expirar antes de fazer cada requisição, não depois que já venceu. Aqui vai uma função que trata isso:

import time

_tokens = {}

def get_valid_token():
    if _tokens and _tokens.get("expires_at", 0) > time.time() + 300:
        return _tokens["access_token"]
    
    data = get_access_token()
    _tokens["access_token"] = data["access_token"]
    _tokens["expires_in"] = data.get("expires_in", 3600)
    _tokens["expires_at"] = time.time() + _tokens["expires_in"]
    if "refresh_token" in data:
        _tokens["refresh_token"] = data["refresh_token"]
    
    return _tokens["access_token"]

O +300 é de cinco minutos. Isso garante que o token seja renovado antes de qualquer requisição que dependa dele falhar. Cinco minutos é margem suficiente pra processamento lento ou retransmissões de rede. Depois que você sabe autenticar, vem o problema mais comum: rate limit. Todos os serviços têm limite de requisições. O Google Sheets API permite 300 requisições por minuto por projeto. O Salesforce tem limites baseados em licença. O Slack é ainda mais restritivo. Se você não tratar isso, seu script vai funcionar perfeitamente por dias e depois quebrar sem motivo aparente. A técnica padrão é usar retry com backoff exponencial. Quando o servidor responde com 429 Too Many Requests ou 503 Service Unavailable, você espera um tempo e tenta de novo. O tempo de espera deve dobrar a cada tentativa. Aqui vai uma implementação robusta:

import time
import random

def make_request(method, url, headers=None, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = requests.request(
                method, url, headers=headers, timeout=30
            )
            
            if response.status_code == 429:
                wait_time = min(2  attempt + random.uniform(0, 1), 60)
                time.sleep(wait_time)
                continue
            
            response.raise_for_status()
            return response.json()
        
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2  attempt + random.uniform(0, 1)
            time.sleep(wait_time)
    
    return None

Note o random.uniform(0, 1) adicionado ao wait_time. Isso é chamado de jitter. Sem ele, se muitos clientes tentarem renovar ao mesmo tempo, todos vão bater no servidor juntos de novo. O jitter espalha as requisições. Parece pequeno, mas já salvou integração que tava caindo todo dia às 9h da manhã. Outro detalhe importante: alguns serviços mandam um header Retry-After com o tempo exato que você deve esperar. Sempre leia esse header antes de usar sua própria lógica de backoff.

Processamento em lote para evitar timeouts

Quando você precisa processar milhares de registros, fazer uma requisição por item é ineficiente e irrita os servidores de quem está do outro lado. A maioria dos APIs oferece endpoints batch que permitem enviar dezenas de operações numa única requisição. Eu tive um caso onde precisava sincronizar 12.000 contatos de um CRM para um sistema de email marketing. Tentando um por um, com rate limit de 100 requisições por minuto, levava cerca de 2 horas. Mudando para batch de 50 itens por requisição, o mesmo processamento caiu para 15 minutos. A diferença é enorme quando essa tarefa roda todo dia. Veja como estruturar um batch:

def batch_upsert_contacts(contacts, batch_size=50):
    results = []
    for i in range(0, len(contacts), batch_size):
        batch = contacts[i:i + batch_size]
        payload = {"operations": [
            {
                "method": "PATCH",
                "path": f"/contacts/{c['id']}",
                "body": c
            } for c in batch
        ]}
        
        response = make_request(
            "POST",
            "https://api.servico.com/v2/batch",
            headers={
                "Authorization": f"Bearer {get_valid_token()}",
                "Content-Type": "application/json"
            },
            json=payload
        )
        results.extend(response.get("results", []))
    
    return results

Se o serviço não tiver batch nativo, você pode simular processamento paralelo usando threads com limitar de concurrency. Mas cuidado com isso. ThreadPoolExecutor com muitos workers pode facilmente bugar o rate limit que você estava evitando. Use semáforos: Esse semaphore limita a 10 requisições simultâneas. Ajuste conforme o limite do serviço. O Slack permite 50 concorrência na API Web. O HubSpot permite bem mais. Consulte a documentação do serviço antes de chutar um número. A parte mais negligenciada é o log. Quando algo dá errado, sem log estruturado você gasta horas tentando descobrir o que aconteceu. Eu comecei a usar logging com contexto estruturado depois que perdi dois dias rastreando um problema que poderia ter sido achado em cinco minutos com os logs certos.

Esses logs de warning com ID e mensagem de erro específica são o que você vai precisar às 3 da manhã quando o deploy quebrou e o stakeholder tá cobrando resposta. Sem eles, você tá adivinhando. Com eles, você cola o ID num ticket e resolve em dez minutos. Um problema real que eu enfrente é quando a execução para no meio do caminho. Digamos que seu script processou 8.000 de 12.000 contatos e aí cai a rede. Sem checkpoint, você perde todo o progresso ou tenta processar de novo o que já foi feito, criando duplicatas. A solução mais simples é um arquivo de estado em JSON que salva o índice do último item processado: Na prática, isso corta o tempo de recovery de uma execução completa de 2 horas pra cerca de 3 minutos, porque só processa o que pendia. A desvantagem é que se o serviço mudou a estrutura dos dados entre as execuções, você pode terminar com registros inconsistentes. Nesse caso, é melhor limpar o state file e rodar do zero.

Se o volume é grande o suficiente, substitua o arquivo JSON por uma tabela simples num banco SQLite. O ganho é que você pode queryar o histórico e identificar padrões de falha. Para menos de 50 mil registros, o arquivo JSON funciona e não exige infraestrutura adicional.

Testes antes de colocar em produção

Nunca suba uma automação sem testar com dados de staging. Eu vi integração de produção sendo configurada direto contra a API real porque alguém achou que "iam testar primeiro". A pessoa atualizou 4.000 registros errados num CRM de clientes ativos. Recuperar aquilo levou seis horas e uma ligação pro suporte técnico. O teste mínimo que você precisa ter:

• Um ambiente sandbox ou staging do serviço que você integra
• Um conjunto de dados de teste com pelo menos 50 registros representando edge cases
• Um script de validação que compara o resultado esperado com o resultado real
• Teste de failover: simule queda de rede e verifique se o retry funciona

Para o teste de validação, algo simples como isso basta: Isso não cobre tudo, mas pega os erros mais comuns. Se passar nisso, rode no sandbox por um dia antes de liberar pra produção. O sandbox não custa nada e evita dor de cabeça. A parte mais importante que ninguém fala: às vezes a solução certa é não automatizar. Se o volume de dados é baixo, se o processo é exceção e não rotina, se o custo de manutenção da automação é maior que o tempo que você gastaria fazendo manualmente, não faça. Eu fiz essa conta certa e deixe de automatizar um processo de aprovação de notas fiscais que envolvia cinco pessoas e acontecia três vezes por semana. Automatizar ia exigir manutenção constante e ainda assim quebrar. Deixar manual resolveu melhor.

O uso da tecnologia faz sentido quando o problema se repete com frequência suficiente, tem padrão previsível e o volume justifica o investimento em desenvolvimento e manutenção. Quando essas condições não se aplicam, automação é esforço desperdiçado.