Procesando webhooks en Python: FastAPI + Redis

Procesamiento confiable de webhooks con FastAPI y Redis Queue

Los webhooks de servicios de pago exigen respuestas rápidas—típicamente en menos de 5 segundos. El procesamiento síncrono en el endpoint provoca tiempos de espera cuando las bases de datos o servicios externos se retrasan. Una implementación ingenua sería esta:

@app.post("/webhook")
async def webhook(request: dict):
    update_database(request)
    send_email_to_user(request)
    return {"status": "ok"}

Este enfoque es frágil: un fallo del servidor durante el procesamiento causa pérdida de datos, y notificaciones duplicadas generan errores de restricción única en la base de datos. La solución? Desacoplar la recepción del procesamiento usando una cola de tareas.

Arquitectura con cola de tareas

El endpoint de la API recibe el webhook, valida la firma, asegura la idempotencia y encola la tarea en Redis. Un trabajador separado ejecuta asíncronamente la lógica de negocio. Las ventajas incluyen:

• Respuesta HTTP instantánea (menos de 50 ms).
• Persistencia de tareas incluso si el trabajador falla.
• Escalabilidad horizontal (varios trabajadores pueden ejecutarse al mismo tiempo).

Componentes:

• FastAPI para la capa de API.
• Redis como cola de mensajes (usando LPUSH y BRPOP).
• Trabajadores que corren en procesos aislados.

Implementación del receptor de API

El endpoint se centra en seguridad y velocidad. Verifica la firma HMAC, evita duplicados mediante transaction_id y agrega la tarea a la cola.

from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
import redis
import json
import hashlib
import os

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
SECRET = os.getenv('WEBHOOK_SECRET', 'test_secret')

def check_signature(body: str, sign: str) -> bool:
    my_sign = hashlib.sha256(f"{body}{SECRET}".encode()).hexdigest()
    return my_sign == sign

@app.post("/api/payment")
async def payment_webhook(request: dict, x_sign: str = Header(None)):
    if not check_signature(json.dumps(request), x_sign):
        raise HTTPException(status_code=403, detail="Bad sign")
    
    task_id = request.get('transaction_id')
    if r.exists(f"processed:{task_id}"):
        return {"status": "duplicate"}
    
    task = {
        "id": task_id,
        "amount": request.get('amount'),
        "user_id": request.get('user_id')
    }
    r.lpush("payment_queue", json.dumps(task))
    
    return {"status": "accepted"}

Prácticas clave:

• La firma HMAC evita solicitudes falsificadas.
• La clave Redis processed:{id} garantiza idempotencia.
• LPUSH añade tareas a la cola de forma eficiente.

Trabajador en segundo plano para procesar tareas

El trabajador se ejecuta en un bucle, extrae tareas con BRPOP (timeout de 5 segundos), ejecuta la lógica y marca la tarea como procesada (con TTL de 1 día). En caso de error, vuelve a intentarlo tras un retraso.

import redis
import json
import time
import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

def process_task(data):
    log.info(f"Processing payment {data['id']}")
    time.sleep(1) 
    return True

def main():
    r = redis.Redis(host='localhost', port=6379)
    log.info("Worker started...")
    
    while True:
        task = r.brpop("payment_queue", timeout=5)
        if not task:
            continue
        _, raw_data = task
        data = json.loads(raw_data)
        
        try:
            success = process_task(data)
            if success:
                r.setex(f"processed:{data['id']}", 86400, "1")
                log.info(f"Task {data['id']} completed")
        except Exception as e:
            log.error(f"Error: {e}")
            r.lpush("payment_queue", raw_data)
            time.sleep(5)

if __name__ == "__main__":
    main()

Problemas comunes y soluciones

• Validación de firma ausente: Vulnerable a POSTs falsos. Solución: Usa HMAC con una clave secreta.
• Volatilidad de Redis: Datos perdidos al reiniciar. Solución: Activa persistencia RDB/AOF o usa PostgreSQL como cola.
• Bucles de reintento del trabajador: Tareas atrapadas en reintentos infinitos. Solución: Controla los intentos, usa colas de mensajes muertos y aplica backoff exponencial.

Otros riesgos:

• Tiempos de espera en servicios externos: Establece tiempos de espera en process_task.
• Dificultades de escalado: Monitorea la longitud de la cola (LLEN) y escala los trabajadores según sea necesario.

Conclusiones clave

• Idempotencia: Claves como processed:{id} evitan procesamientos duplicados.
• Seguridad: Siempre valida firmas HMAC en webhooks entrantes.
• Fiabilidad: Las colas preservan tareas durante fallos.
• Escalabilidad: Trabajadores independientes permiten procesamiento paralelo.
• Monitoreo: Registra métricas de cola y errores para observabilidad.

— Editorial Team