Przetwarzanie webhooków w Pythonie: FastAPI + Redis

Niezawodna obsługa webhooków w FastAPI z kolejką Redis

Webhooki od usług płatności wymagają szybkiej odpowiedzi — zwykle nie więcej niż 5 sekund. Synchroniczna przetwarzanie w punkcie końcowym prowadzi do wygaśnięcia przy opóźnieniach w bazie danych lub zewnętrznych usługach. Przykład początkowej implementacji:

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

Taka architektura jest narażona na ryzyko: awaria serwera podczas przetwarzania prowadzi do utraty danych, powtarzające się powiadomienia wywołują błędy unikalności w bazie danych. Rozwiązaniem jest rozdzielenie odbioru i przetwarzania za pomocą kolejki zadań.

Architektura z kolejką zadań

Punkt końcowy API odbiera webhook, sprawdza podpis, zapewnia idempotentność i umieszcza zadanie w kolejce Redis. Oddzielny worker asynchronicznie wykonuje logikę biznesową. Zalety:

• Natychmiastowa odpowiedź HTTP (50 ms).
• Zapisywanie zadań przy awarii workera.
• Skalowanie poziome (wiele workerów).

Składniki:

• FastAPI do obsługi API.
• Redis jako kolejka (LPUSH/BRPOP).
• Workery w osobnych procesach.

Implementacja odbiornika API

Punkt końcowy skupia się na bezpieczeństwie i szybkości. Sprawdza podpis HMAC, eliminuje duplikaty na podstawie transaction_id, dodaje zadanie do kolejki.

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

Kluczowe praktyki:

• Podpis HMAC chroni przed podszywaniem żądań.
• Klucz Redis processed:{id} zapewnia idempotentność.
• LPUSH do dodawania do kolejki.

Worker do przetwarzania w tle

Worker działa w pętli, pobiera zadania blokującym BRPOP (timeout 5s), wykonuje logikę, oznacza jako przetworzone (TTL 1 dzień). W przypadku błędu — zwraca do kolejki z opóźnieniem.

import redis
import json
import time
import logging

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

def process_task(data):
    log.info(f"Przetwarzamy płatność {data['id']}")
    time.sleep(1) 
    return True

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

if __name__ == "__main__":
    main()

Typowe problemy i rozwiązania

• Brak sprawdzania podpisu: Podatność na fałszywe żądania POST. Rozwiązanie — HMAC z kluczem tajnym.
• Wolność Redis: Dane w pamięci giną przy restartach. Rozwiązanie — trwałe persistencje RDB/AOF lub PostgreSQL jako kolejka.
• Cykl błędów w workere: Zadanie wraca bez końca. Rozwiązanie — licznik prób, dead letter queue, opóźnienia.

Inne ryzyka:

• Timeouty zewnętrznych usług: używać timeoutów w process_task.
• Skalowanie: monitorować długość kolejki (LLEN), dodawać workerów.

Co jest ważne

• Idempotentność: Klucze processed:{id} zapobiegają duplikatom.
• Bezpieczeństwo: Zawsze sprawdzaj podpis HMAC webhooków.
• Niezawodność: Kolejki zachowują zadania przy awariach.
• Skalowalność: Niezależne workery pozwalają na równoległe przetwarzanie.
• Monitorowanie: Rejestruj metryki kolejki i błędów.

— Editorial Team