Webhooks in Python verarbeiten: FastAPI + Redis

Zuverlässige Webhook-Verarbeitung mit FastAPI und Redis Queue

Payment-Service-Webhooks erfordern schnelle Antworten – typischerweise innerhalb von unter fünf Sekunden. Synchrones Verarbeiten im Endpunkt führt bei Datenbank- oder externen Service-Latenzen zu Timeouts. Eine naive Implementierung sieht so aus:

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

Dieser Ansatz ist anfällig: Ein Server-Crash während der Verarbeitung führt zu Datenverlust, und doppelte Benachrichtigungen verursachen Unique-Key-Fehler in der Datenbank. Die Lösung? Empfang und Verarbeitung durch eine Aufgabenwarteschlange entkoppeln.

Architektur mit einer Aufgabenwarteschlange

Der API-Endpunkt empfängt den Webhook, überprüft die Signatur, stellt Idempotenz sicher und fügt die Aufgabe in Redis ein. Ein separater Worker führt die Geschäftslogik asynchron aus. Vorteile sind:

• Sofortige HTTP-Antwort (unter 50 ms).
• Aufgabendauerhaftigkeit auch bei Worker-Crash.
• Horizontale Skalierbarkeit (mehrere Worker laufen gleichzeitig).

Komponenten:

• FastAPI für die API-Schicht.
• Redis als Nachrichtenwarteschlange (mit LPUSH und BRPOP).
• Worker in isolierten Prozessen.

Implementierung des API-Empfängers

Der Endpunkt konzentriert sich auf Sicherheit und Geschwindigkeit. Er prüft die HMAC-Signatur, verhindert Doppelungen über transaction_id und fügt die Aufgabe in die Warteschlange ein.

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="Falsche Signatur")
    
    task_id = request.get('transaction_id')
    if r.exists(f"processed:{task_id}"):
        return {"status": "duplikat"}
    
    task = {
        "id": task_id,
        "amount": request.get('amount'),
        "user_id": request.get('user_id')
    }
    r.lpush("payment_queue", json.dumps(task))
    
    return {"status": "akzeptiert"}

Wichtige Praktiken:

• HMAC-Signatur schützt vor manipulierten Anfragen.
• Redis-Schlüssel processed:{id} gewährleistet Idempotenz.
• LPUSH fügt Aufgaben effizient in die Warteschlange ein.

Hintergrundworker für die Aufgabenverarbeitung

Der Worker läuft in einer Schleife, holt Aufgaben mit BRPOP (5-Sekunden-Timeout), führt die Logik aus und markiert sie als abgeschlossen (mit 1-Tages-TTL). Bei Fehlern wird nach einer Verzögerung erneut versucht.

import redis
import json
import time
import logging

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

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

def main():
    r = redis.Redis(host='localhost', port=6379)
    log.info("Worker gestartet...")
    
    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"Aufgabe {data['id']} abgeschlossen")
        except Exception as e:
            log.error(f"Fehler: {e}")
            r.lpush("payment_queue", raw_data)
            time.sleep(5)

if __name__ == "__main__":
    main()

Häufige Probleme und Lösungen

• Fehlende Signaturüberprüfung: Anfällig für gefälschte POSTs. Lösung: HMAC mit geheimem Schlüssel verwenden.
• Redis-Volatile: Daten gehen beim Neustart verloren. Lösung: RDB/AOF-Persistenz aktivieren oder PostgreSQL als Warteschlange nutzen.
• Worker-Wiederholungsschleifen: Aufgaben hängen in endlosen Wiederholungen fest. Lösung: Versuche zählen, Dead-Letter-Queues nutzen und exponentiellen Backoff implementieren.

Weitere Risiken:

• Timeout externer Services: Timeout in process_task setzen.
• Skalierungsprobleme: Warteschlangenlänge (LLEN) überwachen und Worker entsprechend skalieren.

Wichtige Erkenntnisse

• Idempotenz: Schlüssel wie processed:{id} verhindern doppelte Verarbeitung.
• Sicherheit: HMAC-Signaturen bei eingehenden Webhooks immer validieren.
• Zuverlässigkeit: Warteschlangen bewahren Aufgaben auch bei Ausfällen.
• Skalierbarkeit: Unabhängige Worker ermöglichen parallele Verarbeitung.
• Überwachung: Warteschlangen-Metriken und Fehler protokollieren für Observability.

— Editorial Team