Trabalhando com APIs e Webhooks
Scripts de automação integram-se com APIs SaaS e internas através de clientes HTTP. Use httpx ou requests com timeouts, retentativas, loops de paginação e receptores de webhook opcionais para gatilhos orientados a eventos.
Receita
import httpx
with httpx.Client(timeout=30.0) as client:
resp = client.get(
"https://api.example.com/v1/items",
headers={"Authorization": "Bearer TOKEN"},
params={"page": 1, "limit": 100},
)
resp.raise_for_status()
data = resp.json()Quando usar isso:
- Polling quando o fornecedor não possui webhooks
- Scripts handler de webhook atrás de proxy reverso ou Lambda
- APIs de exportação em massa com paginação por cursor
- APIs admin internas para automação de operações
Exemplo de Trabalho
Cliente GET paginado com retentativa, mais um stub mínimo de verificação de assinatura de webhook.
import hashlib
import hmac
import time
from typing import Any, Iterator
import httpx
class ApiClient:
def __init__(self, base_url: str, token: str) -> None:
self._client = httpx.Client(
base_url=base_url,
headers={"Authorization": f"Bearer {token}"},
timeout=httpx.Timeout(30.0),
)
def list_items(self, page_size: int = 100) -> Iterator[dict[str, Any]]:
cursor: str | None = None
while True:
params = {"limit": page_size}
if cursor:
params["cursor"] = cursor
resp = self._client.get("/v1/items", params=params)
resp.raise_for_status()
payload = resp.json()
for item in payload.get("items", []):
yield item
cursor = payload.get("next_cursor")
if not cursor:
break
def close(self) -> None:
self._client.close()
def verify_webhook(body: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
if __name__ == "__main__":
client = ApiClient("https://api.example.com", "dev-token")
for i, item in enumerate(client.list_items()):
if i >= 3:
break
print(item.get("id"))
client.close()O que isso demonstra:
- Paginação por cursor até
next_cursorestar ausente httpx.Clientcompartilhado reutiliza conexões entre páginashmac.compare_digestprevine vazamentos de tempo na verificação de webhook
Mergulho Profundo
Polling vs Webhooks
| Modo | Prós | Contras |
|---|---|---|
| Polling | Simples, sem porta de entrada | Atraso, limites de taxa |
| Webhooks | Tempo real | Verificação de assinatura, retentativas, idempotência |
Resiliência
- Timeouts na conexão e leitura
- Retentar GETs idempotentes em 502/503 com backoff
- Respeitar o cabeçalho
Retry-Afterem 429
Notas de Python
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=0.5))
def fetch(client: httpx.Client, url: str) -> httpx.Response:
resp = client.get(url)
resp.raise_for_status()
return respArmadilhas
- Sem timeout - scripts travam para sempre. Correção:
httpx.Timeoutsempre definido. - Confiar no corpo do webhook sem verificação - eventos forjados. Correção: HMAC ou mTLS conforme documentação do fornecedor.
- Não idempotente em retentativas de webhook - linhas duplicadas. Correção: armazenar IDs de eventos processados.
- Carregar todas as páginas na RAM - OOM em exportações grandes. Correção: padrão de gerador processando por página.
- Segredos em parâmetros de consulta de URL - registrados por proxies. Correção: apenas cabeçalho Authorization.
Alternativas
| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
| SDK Oficial | Fornecedor fornece SDK Python mantido | Superfície REST fina |
| SQS/Lambda webhook | Escalar eventos de entrada | Um único poller cron é suficiente |
| Cliente GraphQL | API é apenas GraphQL | Paginação REST simples |
FAQs
httpx vs requests?
httpx suporta HTTP/2 e async; requests é apenas síncrono, mas ubíquo - ambos servem para scripts com timeouts.
Como lidar com limites de taxa?
Ler cabeçalhos X-RateLimit-*, dormir ou fazer backoff em 429, reduzir concorrência.
Paginação por offset vs cursor?
Preferir cursor para grandes conjuntos de dados - paginação por offset falha quando os dados mutam durante a navegação.
Como executar um receptor de webhook?
Endpoint FastAPI/Flask ou URL de Função Lambda - validar assinatura antes do trabalho.
Como testar scripts HTTP?
httpx.MockTransport ou fixtures pytest-httpx - sem rede em testes unitários.
Scripts devem usar httpx async?
Síncrono é mais simples para cron; async ao orquestrar muitos downloads paralelos em um único loop de eventos.
Como registrar requisições?
Registrar método, caminho da URL, status, duração - não o cabeçalho Authorization ou corpos PII completos.
E os links de paginação?
Algumas APIs usam o cabeçalho RFC 5988 Link - analisar rel="next" em vez do campo cursor JSON.
Como passar JSON POST?
client.post("/hooks", json={"ok": True}) define o content-type automaticamente.
Como isso se relaciona com agendamento?
Scripts baseados em polling são frequentemente acionados por cron; webhooks reduzem a frequência de agendamento.
Relacionados
- Noções Básicas de Scripting - CLI e configuração de ambiente
- Agendamento e Cron - agendamentos de polling
- Scripts Robustos - retentativas e idempotência
- httpx & requests - Profundidade do cliente HTTP
- Melhores Práticas de Scripting de Sistema - regras de automação de API
Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.