Trabajar con APIs y Webhooks
Los scripts de automatización se integran con APIs SaaS e internas a través de clientes HTTP. Utiliza httpx o requests con tiempos de espera, reintentos, bucles de paginación y receptores de webhook opcionales para disparadores basados en eventos.
Receta
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()Cuándo usar esto:
- Polling cuando el proveedor carece de webhooks
- Scripts de manejador de webhooks detrás de un proxy inverso o Lambda
- APIs de exportación masiva con paginación por cursor
- APIs de administración interna para automatización de operaciones
Ejemplo de Trabajo
Cliente GET paginado con reintento, además de un stub mínimo de verificación de firma 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()Lo que esto demuestra:
- Paginación por cursor hasta que
next_cursoresté ausente httpx.Clientcompartido reutiliza conexiones entre páginashmac.compare_digestpreviene fugas de tiempo en la verificación de webhooks
Inmersión Profunda
Polling vs Webhooks
| Modo | Pros | Contras |
|---|---|---|
| Polling | Simple, sin puerto de entrada | Retraso, límites de tasa |
| Webhooks | Tiempo real | Verificación de firma, reintentos, idempotencia |
Resiliencia
- Tiempos de espera de conexión y lectura
- Reintentar GETs idempotentes en 502/503 con retroceso (backoff)
- Respetar la cabecera
Retry-Afteren 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 respErrores Comunes
- Sin tiempo de espera - scripts colgados para siempre. Solución:
httpx.Timeoutsiempre establecido. - Confiar en el cuerpo del webhook sin verificar - eventos forjados. Solución: HMAC o mTLS según la documentación del proveedor.
- No idempotente en reintentos de webhook - filas duplicadas. Solución: almacenar IDs de eventos procesados.
- Cargar todas las páginas en RAM - OOM (Error de Memoria) en exportaciones grandes. Solución: patrón generador procesando por página.
- Secretos en parámetros de consulta de URL - registrados por proxies. Solución: solo cabecera de Autorización.
Alternativas
| Alternativa | Usar Cuando | No Usar Cuando |
|---|---|---|
| SDK Oficial | El proveedor proporciona un SDK de Python mantenido | Superficie REST delgada |
| SQS/Lambda webhook | Escalar eventos entrantes | Un solo poller cron es suficiente |
| Cliente GraphQL | La API es solo GraphQL | Paginación REST simple |
Preguntas Frecuentes
¿httpx vs requests?
httpx soporta HTTP/2 y asíncrono; requests es solo síncrono pero ubicuo - ambos están bien para scripts con tiempos de espera.
¿Cómo manejo los límites de tasa?
Leer las cabeceras X-RateLimit-*, esperar o retroceder en 429, reducir la concurrencia.
¿Paginación por offset vs cursor?
Preferir cursor para grandes conjuntos de datos - la paginación por offset falla cuando los datos mutan durante el recorrido.
¿Cómo ejecuto un receptor de webhook?
Endpoint de FastAPI/Flask o URL de Función Lambda - validar la firma antes de trabajar.
¿Cómo pruebo scripts HTTP?
httpx.MockTransport o fixtures de pytest-httpx - sin red en pruebas unitarias.
¿Deberían los scripts usar httpx asíncrono?
Síncrono es más simple para cron; asíncrono al orquestar muchas descargas paralelas en un único bucle de eventos.
¿Cómo registro las solicitudes?
Registrar método, ruta de URL, estado, duración - no la cabecera de Autorización ni los cuerpos completos de PII.
¿Qué pasa con los enlaces de paginación?
Algunas APIs usan la cabecera Link de RFC 5988 - analizar rel="next" en lugar del campo de cursor JSON.
¿Cómo paso JSON en POST?
client.post("/hooks", json={"ok": True}) establece el content-type automáticamente.
¿Cómo se relaciona esto con la programación?
Los scripts basados en polling a menudo se ejecutan con cron; los webhooks reducen la frecuencia de programación.
Relacionado
- Conceptos Básicos de Scripting - CLI y configuración de entorno
- Programación y Cron - horarios de polling
- Scripts Robustos - reintentos e idempotencia
- httpx y requests - profundidad del cliente HTTP
- Mejores Prácticas de Scripting de Sistemas - reglas de automatización de API
Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+, y uv 0.6+.