Traitement des webhooks en Python : FastAPI + Redis

Traitement fiable des webhooks avec FastAPI et Redis Queue

Les webhooks de service de paiement exigent des réponses rapides — généralement en moins de 5 secondes. Un traitement synchrone dans l'endpoint entraîne des timeouts lorsque la base de données ou les services externes ralentissent. Une implémentation naïve ressemble à ceci :

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

Cette approche est fragile : une panne du serveur pendant le traitement entraîne une perte de données, et les notifications en double provoquent des erreurs de contrainte unique dans la base. La solution ? Découpler la réception du traitement grâce à une file d'attente de tâches.

Architecture avec une file d'attente de tâches

L'endpoint API reçoit le webhook, vérifie la signature, assure l'idempotence, puis ajoute la tâche à Redis. Un worker distinct exécute asynchrone la logique métier. Les avantages incluent :

• Réponse HTTP instantanée (moins de 50 ms).
• Persistance des tâches même si le worker plante.
• Scalabilité horizontale (plusieurs workers peuvent fonctionner simultanément).

Composants :

• FastAPI pour la couche API.
• Redis comme file de messages (avec LPUSH et BRPOP).
• Workers exécutés dans des processus isolés.

Implémentation du récepteur API

L'endpoint se concentre sur la sécurité et la rapidité. Il vérifie la signature HMAC, empêche les doublons via transaction_id, et ajoute la tâche à la file.

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"}

Bonnes pratiques :

• La signature HMAC empêche les requêtes falsifiées.
• La clé Redis processed:{id} garantit l'idempotence.
• LPUSH ajoute efficacement les tâches à la file.

Worker en arrière-plan pour le traitement des tâches

Le worker tourne en boucle, récupère les tâches avec BRPOP (timeout de 5 secondes), exécute la logique, et les marque comme traitées (TTL de 1 jour). En cas d'échec, il retente après un délai.

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()

Problèmes courants et solutions

• Vérification de signature manquante : vulnérable aux faux POSTs. Solution : utiliser HMAC avec une clé secrète.
• Volatilité de Redis : les données sont perdues au redémarrage. Solution : activer la persistance RDB/AOF ou utiliser PostgreSQL comme file.
• Boucles de réessai du worker : les tâches bloquent en boucle infinie. Solution : compter les tentatives, utiliser des files mortes (dead letter queues), et appliquer un backoff exponentiel.

Autres risques :

• Temps d’attente des services externes : définir des timeouts dans process_task.
• Difficultés d’évolutivité : surveiller la longueur de la file (LLEN) et ajuster le nombre de workers.

Points clés

• Idempotence : des clés comme processed:{id} empêchent le traitement en double.
• Sécurité : toujours valider les signatures HMAC sur les webhooks entrants.
• Fiabilité : les files conservent les tâches en cas d’échec.
• Scalabilité : les workers indépendants permettent un traitement parallèle.
• Surveillance : journaliser les métriques de file et les erreurs pour une meilleure observabilité.

— Editorial Team