Processing webhooks in Python: FastAPI + Redis

Reliable Webhook Processing with FastAPI and Redis Queue

Payment service webhooks demand quick responses—typically under 5 seconds. Synchronous processing in the endpoint leads to timeouts when databases or external services lag. A naive implementation looks like this:

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

This approach is fragile: a server crash during processing causes data loss, and duplicate notifications trigger unique constraint errors in the database. The solution? Decouple receiving from processing using a task queue.

Architecture with a Task Queue

The API endpoint receives the webhook, validates the signature, ensures idempotency, and enqueues the task in Redis. A separate worker asynchronously executes the business logic. Benefits include:

• Instant HTTP response (under 50ms).
• Task persistence even if the worker crashes.
• Horizontal scalability (multiple workers can run simultaneously).

Components:

• FastAPI for the API layer.
• Redis as the message queue (using LPUSH and BRPOP).
• Workers running in isolated processes.

Implementing the API Receiver

The endpoint focuses on security and speed. It checks the HMAC signature, prevents duplicates via transaction_id, and adds the task to the queue.

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

Key practices:

• HMAC signature prevents forged requests.
• Redis key processed:{id} ensures idempotency.
• LPUSH adds tasks to the queue efficiently.

Background Worker for Task Processing

The worker runs in a loop, pulling tasks with BRPOP (5-second timeout), executing the logic, and marking them as processed (with a 1-day TTL). On failure, it retries after a delay.

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

Common Issues and Fixes

• Missing signature validation: Vulnerable to fake POSTs. Fix: Use HMAC with a secret key.
• Redis volatility: Data lost on restart. Solution: Enable RDB/AOF persistence or use PostgreSQL as the queue.
• Worker retry loops: Tasks get stuck in endless retries. Fix: Track attempts, use dead letter queues, and implement exponential backoff.

Other risks:

• External service timeouts: Set timeouts in process_task.
• Scaling challenges: Monitor queue length (LLEN) and scale workers accordingly.

Key Takeaways

• Idempotency: Keys like processed:{id} prevent duplicate processing.
• Security: Always validate HMAC signatures on incoming webhooks.
• Reliability: Queues preserve tasks during failures.
• Scalability: Independent workers allow parallel processing.
• Monitoring: Log queue metrics and errors for observability.

— Editorial Team