FastAPI Webhooks in the Real World: Signatures, Idempotency, and Background Processing

FastAPI Webhooks in the Real World: Signatures, Idempotency, and Background Processing

Webhooks look simple: an external service sends you an HTTP POST, you respond 200, done. In production, that “simple” endpoint becomes a reliability hotspot: you must verify the sender, avoid double-processing, return quickly (or the sender retries), and handle bursts without taking your app down.

This hands-on guide shows a practical pattern for building a robust webhook receiver in FastAPI with:

• Signature verification (HMAC) to authenticate requests
• Idempotency to prevent duplicate processing
• Quick ACK responses while processing in the background
• Safe logging and structured error handling
• Minimal persistence using SQLite (works locally; swap later)

Project setup

Install dependencies:

pip install fastapi uvicorn pydantic sqlalchemy

We’ll use a tiny SQLite database to track processed events (idempotency). Here’s a simple structure:

app/ main.py db.py models.py security.py worker.py

1) Database for idempotency (SQLite + SQLAlchemy)

We store each webhook’s unique event ID (or a hash if the provider doesn’t give one). If we’ve seen it before, we skip processing.

# app/db.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL = "sqlite:///./webhooks.db" engine = create_engine( DATABASE_URL, connect_args={"check_same_thread": False}, # required for SQLite + threads ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def get_db(): db = SessionLocal() try: yield db finally: db.close() 

# app/models.py from sqlalchemy import Column, String, DateTime from sqlalchemy.sql import func from .db import Base class WebhookEvent(Base): __tablename__ = "webhook_events" event_id = Column(String, primary_key=True, index=True) received_at = Column(DateTime(timezone=True), server_default=func.now()) 

2) Signature verification (HMAC) you can actually trust

Many providers sign the raw request body with a shared secret and include the signature in a header (for example X-Signature). The important part is verifying the raw bytes (not the parsed JSON) using a constant-time compare.

# app/security.py import hmac import hashlib def compute_hmac_sha256(secret: str, payload: bytes) -> str: mac = hmac.new(secret.encode("utf-8"), payload, hashlib.sha256) return mac.hexdigest() def verify_signature(secret: str, payload: bytes, provided_signature: str) -> bool: expected = compute_hmac_sha256(secret, payload) # Constant-time compare prevents timing attacks return hmac.compare_digest(expected, provided_signature.strip().lower()) 

Tip: Some services prefix the signature (e.g., sha256=...). Normalize it before comparing.

3) Background processing without blocking the webhook

Your webhook endpoint should respond quickly (often within a few seconds). If you do slow work inline (calling other APIs, heavy DB writes), the provider retries, causing duplicates and load spikes.

FastAPI includes BackgroundTasks, which is perfect for lightweight background work. For heavy work, use a proper queue (Celery/RQ/Arq), but the pattern is the same.

# app/worker.py import time def process_event(event_type: str, data: dict) -> None: """ Replace this with real work: - update your database - call internal services - enqueue a real job, etc. """ # Simulate slow processing time.sleep(1.0) if event_type == "user.created": # Example: do something with data user_id = data.get("id") email = data.get("email") print(f"[worker] created user id={user_id} email={email}") else: print(f"[worker] ignoring event_type={event_type}") 

4) The webhook endpoint (verify, dedupe, ACK fast)

Now we combine everything in one practical endpoint:

• Read raw body
• Verify signature
• Extract event_id (or compute one)
• Check/insert idempotency record
• Queue processing in the background
• Return 202 Accepted (or 200 OK) quickly

# app/main.py import hashlib from fastapi import FastAPI, Request, Header, HTTPException, BackgroundTasks, Depends from sqlalchemy.orm import Session from .db import Base, engine, get_db from .models import WebhookEvent from .security import verify_signature from .worker import process_event WEBHOOK_SECRET = "change-me-in-env" # use env vars in real apps app = FastAPI(title="Webhook Receiver") Base.metadata.create_all(bind=engine) def compute_event_id(payload: bytes) -> str: """ Fallback idempotency key when no provider event ID exists. Using a hash of raw body is a common approach. """ return hashlib.sha256(payload).hexdigest() @app.post("/webhooks/provider") async def provider_webhook( request: Request, background: BackgroundTasks, db: Session = Depends(get_db), x_signature: str | None = Header(default=None, alias="X-Signature"), x_event_id: str | None = Header(default=None, alias="X-Event-Id"), x_event_type: str | None = Header(default=None, alias="X-Event-Type"), ): raw = await request.body() # 1) Verify signature if not x_signature: raise HTTPException(status_code=401, detail="Missing signature header") if not verify_signature(WEBHOOK_SECRET, raw, x_signature): raise HTTPException(status_code=401, detail="Invalid signature") # 2) Parse JSON after verifying authenticity try: payload = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON") event_type = x_event_type or payload.get("type") or "unknown" event_id = x_event_id or payload.get("id") or compute_event_id(raw) # 3) Idempotency check (dedupe) existing = db.get(WebhookEvent, event_id) if existing: # Already processed (or in progress). ACK quickly. return {"status": "duplicate", "event_id": event_id} db.add(WebhookEvent(event_id=event_id)) db.commit() # 4) Background processing background.add_task(process_event, event_type, payload) # 5) Quick ACK return {"status": "accepted", "event_id": event_id, "event_type": event_type} 

Why this order matters: verify signature first (on raw body), then parse JSON. If you parse first, you risk accepting tampered payloads or signature mismatches caused by re-serialization differences.

5) Try it locally with a reproducible request

Run the server:

uvicorn app.main:app --reload

Generate a signature and send a test webhook. This Python snippet simulates a provider:

# send_test_webhook.py import json import hmac import hashlib import requests secret = "change-me-in-env" payload = { "id": "evt_123", "type": "user.created", "data": {"id": "u_42", "email": "[email protected]"}, } raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") sig = hmac.new(secret.encode("utf-8"), raw, hashlib.sha256).hexdigest() res = requests.post( "http://127.0.0.1:8000/webhooks/provider", data=raw, headers={ "Content-Type": "application/json", "X-Signature": sig, "X-Event-Id": payload["id"], "X-Event-Type": payload["type"], }, timeout=5, ) print(res.status_code, res.json()) 

Run it twice. The first call should return accepted, the second should return duplicate. That’s idempotency working.

6) Production notes junior devs usually miss

• Use environment variables for WEBHOOK_SECRET. Never hardcode secrets.
• Return quickly. Even if you don’t use background tasks yet, keep the endpoint fast to reduce retries.
• Handle provider retries. Retries are normal (timeouts, 5xx, network hiccups). Idempotency isn’t optional.
• Log safely. Don’t dump full payloads with PII. Log event_id, event_type, and minimal metadata.
• Think about “in progress”. This example marks an event as “seen” before processing. If processing fails, you may need a status column (pending/done/failed) and a retry mechanism.
• Validate schema. After verifying the signature, validate payload fields (Pydantic models help a lot) so broken payloads don’t crash your worker.

7) A small upgrade: persist status for retries

If you want a more production-friendly approach, extend WebhookEvent with a status column and update it after processing. Then you can reprocess failed events with a simple admin script.

Even if you don’t implement that today, this article’s pattern (verify → dedupe → queue → ACK) will save you from the most common webhook failures.

Wrap-up

A reliable webhook receiver isn’t about fancy architecture—it’s about doing the fundamentals every time: authenticate the sender, dedupe aggressively, and respond fast. With FastAPI, you can get a clean baseline in a few files, and later swap the background task for a real queue and SQLite for Postgres without changing the core flow.