Webhook ingestion sounds simple — receive an HTTP POST, write the data — until you're debugging dropped events at 2am. This lesson teaches you to build a production-grade webhook pipeline with cryptographic validation, durable queuing, idempotent processing, and replay tooling.
Your Stripe account just processed a payment. A customer updated their profile in your CRM. A GitHub Actions workflow completed a deployment. In each of these moments, something happened in an external system — and your data pipeline needs to know about it, ideally within seconds rather than hours.
This is the core promise of webhook-based ingestion: instead of your pipeline waking up on a schedule and asking "did anything happen?", the source system tells you the moment something does. But the gap between receiving a webhook and reliably processing it in production is where most implementations fall apart. Teams bolt together a Flask endpoint, write the event directly to the database, and discover too late that they've built something that drops data under load, has no replay capability, and can be spoofed by anyone who knows the URL.
By the end of this lesson, you'll know how to build a webhook ingestion layer that's production-worthy: one that validates incoming requests cryptographically, queues events durably before processing, handles retries and deduplication correctly, and fails gracefully when downstream systems are unavailable. We'll build a real example around Stripe payment webhooks feeding an analytics pipeline, but every pattern here applies equally to GitHub, Shopify, Segment, PagerDuty, or any other system that speaks HTTP callbacks.
What you'll learn:
You should be comfortable with Python and have a working understanding of REST APIs — if you need a refresher on HTTP mechanics, working with APIs for data engineers covers the fundamentals. You should also understand the basics of queue-based messaging (Redis, RabbitMQ, or SQS conceptually) and have Python 3.10+, FastAPI, and Redis available in your development environment. Familiarity with batch vs. stream processing tradeoffs will help you contextualize where webhook ingestion fits.
Before writing any code, it's worth being precise about the contract a webhook creates. When a source system fires a webhook, it makes an HTTP POST to your endpoint and waits for a response. Most systems (Stripe included) expect a 2xx response within 5–30 seconds. If they don't get one, they retry — often with exponential backoff, sometimes for 24–72 hours.
This creates an asymmetry that breaks naive implementations: the sender's retry logic is your recovery mechanism, but only if you use it correctly. The typical mistake looks like this:
# DON'T DO THIS — the naive pattern
@app.post("/webhooks/stripe")
async def handle_stripe_webhook(request: Request):
payload = await request.json()
# This can take 2-30 seconds and fail halfway through
process_payment_event(payload)
write_to_warehouse(payload)
update_dashboard_cache(payload)
return {"status": "ok"}
The problem: you're doing expensive, fallible work synchronously before acknowledging receipt. If write_to_warehouse takes 15 seconds, you'll timeout. If it raises an exception, you return a 5xx and Stripe retries — potentially processing the same event again. If your service restarts mid-processing, the event is gone.
The correct mental model has two distinct phases:
This separation is the foundation of everything that follows.
Key insight
Your webhook endpoint's only job is to be a fast, reliable mailbox. Processing is someone else's problem — specifically, a worker that reads from a queue. This decoupling is what lets you handle traffic spikes, survive worker crashes, and replay events without asking the sender to retransmit.
Before you persist anything, you need to verify that the request actually came from your source system and wasn't tampered with in transit. Stripe, GitHub, Shopify, and most modern webhook senders include an HMAC signature in the request headers. Here's how it works and how to verify it correctly.
Stripe sends a Stripe-Signature header containing a timestamp and a hash:
Stripe-Signature: t=1699523478,v1=5257a869e7ecebeda32affa62cdca3fa51cad7e77a05bd58b96b74b6f82b94ca
The hash is HMAC-SHA256(timestamp + "." + raw_body, webhook_secret). The timestamp is included in the signed payload to prevent replay attacks — if someone captures a valid request and resends it later, the timestamp check will catch it.
Here's a complete, correct implementation:
import hashlib
import hmac
import time
from fastapi import FastAPI, Request, HTTPException, Header
from typing import Optional
STRIPE_WEBHOOK_SECRET = "whsec_your_secret_here" # Load from secrets manager
STRIPE_TIMESTAMP_TOLERANCE_SECONDS = 300 # 5 minutes
def verify_stripe_signature(
raw_body: bytes,
signature_header: str,
secret: str,
tolerance: int = STRIPE_TIMESTAMP_TOLERANCE_SECONDS
) -> dict:
"""
Validates a Stripe webhook signature.
Returns parsed signature components on success, raises ValueError on failure.
"""
# Parse the header: "t=1699523478,v1=abc123..."
try:
parts = dict(item.split("=", 1) for item in signature_header.split(","))
timestamp = int(parts["t"])
signatures = [v for k, v in parts.items() if k == "v1"]
except (KeyError, ValueError) as e:
raise ValueError(f"Malformed Stripe-Signature header: {e}")
# Check timestamp to prevent replay attacks
current_time = int(time.time())
if abs(current_time - timestamp) > tolerance:
raise ValueError(
f"Webhook timestamp {timestamp} is outside tolerance window "
f"({tolerance}s). Possible replay attack."
)
# Compute expected signature
signed_payload = f"{timestamp}.{raw_body.decode('utf-8')}"
expected_sig = hmac.new(
secret.encode("utf-8"),
signed_payload.encode("utf-8"),
hashlib.sha256
).hexdigest()
# Constant-time comparison to prevent timing attacks
if not any(hmac.compare_digest(expected_sig, sig) for sig in signatures):
raise ValueError("Signature mismatch — request may be forged")
return {"timestamp": timestamp, "signatures": signatures}
Two subtleties here deserve attention. First, we use hmac.compare_digest() rather than == for string comparison. Regular string equality short-circuits on the first mismatched character, which creates a timing side channel — an attacker can statistically determine how many characters of a valid signature they've guessed. compare_digest always takes the same time regardless of where the mismatch occurs.
Second, notice we read the raw body as bytes before parsing it. The signature is computed over the exact bytes sent, not a re-serialized version of the parsed JSON. If you call await request.json() first, then try to reconstruct bytes from the parsed dict, you'll get different bytes (key ordering, whitespace) and the signature won't match.
Warning
Never load your webhook secret from an environment variable hardcoded in source code. Use a dedicated secrets manager. The secrets management guide for data pipelines covers the right patterns for loading credentials at runtime without exposing them in logs or version control.
Now let's build the complete receiver. The key design choices:
import json
import uuid
import redis
import structlog
from fastapi import FastAPI, Request, HTTPException
from datetime import datetime, timezone
app = FastAPI()
log = structlog.get_logger()
# Redis connection — use a connection pool in production
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
WEBHOOK_QUEUE_KEY = "webhooks:stripe:pending"
WEBHOOK_SEEN_KEY = "webhooks:stripe:seen" # For deduplication
@app.post("/webhooks/stripe")
async def receive_stripe_webhook(
request: Request,
stripe_signature: str = Header(None, alias="Stripe-Signature")
):
# Step 1: Read raw body — must happen before any parsing
raw_body = await request.body()
# Step 2: Validate signature
if not stripe_signature:
raise HTTPException(status_code=400, detail="Missing Stripe-Signature header")
try:
verify_stripe_signature(raw_body, stripe_signature, STRIPE_WEBHOOK_SECRET)
except ValueError as e:
log.warning("webhook.signature_invalid", error=str(e),
ip=request.client.host)
raise HTTPException(status_code=400, detail="Invalid signature")
# Step 3: Parse payload and extract event ID
try:
payload = json.loads(raw_body)
except json.JSONDecodeError as e:
raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}")
# Stripe provides a stable event ID — use it for deduplication
event_id = payload.get("id") or str(uuid.uuid4())
event_type = payload.get("type", "unknown")
# Step 4: Deduplication check — has this event_id been received before?
dedup_key = f"{WEBHOOK_SEEN_KEY}:{event_id}"
if redis_client.exists(dedup_key):
log.info("webhook.duplicate_skipped", event_id=event_id, event_type=event_type)
return {"status": "ok", "note": "duplicate"} # Still return 200
# Step 5: Wrap with ingestion metadata
envelope = {
"event_id": event_id,
"event_type": event_type,
"received_at": datetime.now(timezone.utc).isoformat(),
"source": "stripe",
"raw_payload": payload,
"retry_count": 0
}
# Step 6: Enqueue for processing
try:
redis_client.rpush(WEBHOOK_QUEUE_KEY, json.dumps(envelope))
# Mark as seen with TTL to prevent dedup set from growing unbounded
# 48 hours covers Stripe's retry window
redis_client.setex(dedup_key, 172800, "1")
log.info("webhook.received", event_id=event_id, event_type=event_type)
except redis.RedisError as e:
# Critical: queue is unavailable. Log with full context for alerting.
log.error("webhook.queue_unavailable", event_id=event_id,
event_type=event_type, error=str(e))
# Return 500 so Stripe will retry — we haven't persisted this event
raise HTTPException(status_code=500, detail="Queue unavailable")
return {"status": "ok"}
Notice the deduplication pattern at step 4. Stripe guarantees each event has a stable id field. We record this ID in Redis with a 48-hour TTL (matching Stripe's maximum retry window). If the same event arrives twice — which happens legitimately when Stripe retries due to a transient network failure — we acknowledge it and skip re-queuing. The sender gets a 200, nobody gets duplicate data.
Tip
The Redis TTL on your deduplication set is important. Too short and you won't catch retries; too long and the set grows unbounded. Set it to at least as long as your webhook sender's maximum retry window, plus a safety buffer. Stripe retries for 72 hours, so 96 hours is a reasonable TTL.
The worker is a separate process that reads from the queue, applies your pipeline logic, and handles failures. This is where idempotent pipeline design becomes critical — because workers can crash and restart, you may process the same event more than once even with the receiver-side deduplication in place.
import json
import time
import signal
import redis
import structlog
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timezone
log = structlog.get_logger()
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
WEBHOOK_QUEUE_KEY = "webhooks:stripe:pending"
DEAD_LETTER_KEY = "webhooks:stripe:dead_letter"
MAX_RETRIES = 5
PROCESSING_TIMEOUT_SECONDS = 30
@dataclass
class WebhookEnvelope:
event_id: str
event_type: str
received_at: str
source: str
raw_payload: dict
retry_count: int
def process_payment_succeeded(envelope: WebhookEnvelope) -> None:
"""Process a payment_intent.succeeded event."""
payload = envelope.raw_payload
data = payload["data"]["object"]
# Extract fields we care about for analytics
record = {
"event_id": envelope.event_id,
"payment_intent_id": data["id"],
"amount_cents": data["amount"],
"currency": data["currency"],
"customer_id": data.get("customer"),
"metadata": data.get("metadata", {}),
"processed_at": datetime.now(timezone.utc).isoformat(),
"event_timestamp": payload["created"]
}
# In production: write to warehouse, update materialized views, etc.
write_to_warehouse("payment_events", record)
log.info("payment.processed", payment_intent_id=data["id"],
amount_cents=data["amount"])
def route_event(envelope: WebhookEnvelope) -> None:
"""Route events to the appropriate handler based on event type."""
handlers = {
"payment_intent.succeeded": process_payment_succeeded,
"payment_intent.payment_failed": process_payment_failed,
"customer.subscription.updated": process_subscription_update,
# Add handlers as you expand coverage
}
handler = handlers.get(envelope.event_type)
if handler is None:
log.info("webhook.unhandled_event_type", event_type=envelope.event_type,
event_id=envelope.event_id)
return # Not an error — we just don't handle this event type yet
handler(envelope)
def send_to_dead_letter(envelope: WebhookEnvelope, error: str) -> None:
"""Route a failed event to the dead-letter queue for manual inspection."""
dead_letter_record = {
**envelope.__dict__,
"failed_at": datetime.now(timezone.utc).isoformat(),
"final_error": error
}
redis_client.rpush(DEAD_LETTER_KEY, json.dumps(dead_letter_record))
log.error("webhook.dead_lettered", event_id=envelope.event_id,
event_type=envelope.event_type, error=error,
retry_count=envelope.retry_count)
def run_worker():
"""Main worker loop — reads from queue and processes events."""
log.info("webhook_worker.started", queue=WEBHOOK_QUEUE_KEY)
running = True
def handle_shutdown(signum, frame):
nonlocal running
log.info("webhook_worker.shutdown_requested")
running = False
signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown)
while running:
# BLPOP blocks for up to 2 seconds, then returns None
# This allows graceful shutdown without busy-waiting
result = redis_client.blpop(WEBHOOK_QUEUE_KEY, timeout=2)
if result is None:
continue # Timeout — check if still running and loop
_, raw_message = result
try:
data = json.loads(raw_message)
envelope = WebhookEnvelope(**data)
except (json.JSONDecodeError, TypeError) as e:
log.error("webhook_worker.malformed_message", raw=raw_message[:200], error=str(e))
continue
try:
route_event(envelope)
except Exception as e:
envelope.retry_count += 1
if envelope.retry_count >= MAX_RETRIES:
send_to_dead_letter(envelope, str(e))
else:
# Re-queue with incremented retry count and backoff
backoff = min(2 ** envelope.retry_count, 60)
log.warning("webhook_worker.retry_scheduled",
event_id=envelope.event_id,
retry_count=envelope.retry_count,
backoff_seconds=backoff)
time.sleep(backoff)
redis_client.rpush(WEBHOOK_QUEUE_KEY, json.dumps(envelope.__dict__))
if __name__ == "__main__":
run_worker()
The retry logic here uses exponential backoff capped at 60 seconds. After MAX_RETRIES attempts, events go to a dead-letter queue rather than being silently dropped. The dead-letter queue pattern gives you a safe place to inspect, debug, and replay events that repeatedly fail — without blocking the rest of the queue.
Note
The BLPOP command is a blocking list pop in Redis. It's much more efficient than polling with LPOP in a tight loop, but it also means your worker holds an open connection to Redis. Make sure your Redis instance has enough maxclients configured for your worker count, and use connection pooling if you run many workers.
Even with deduplication in the receiver, processing workers can fail after partial execution. The database write might succeed but the cache update might fail, leaving your system in an inconsistent state. When the worker retries, you'll run the handler again. Your handlers need to be idempotent — running them twice should produce the same result as running them once.
The pattern is to use the event_id as a natural idempotency key in your write operations:
def write_to_warehouse(table: str, record: dict) -> None:
"""
Idempotent write using event_id as the conflict resolution key.
Uses upsert semantics so re-processing doesn't create duplicate rows.
"""
import psycopg2
conn = get_db_connection() # Use connection pooling in production
with conn.cursor() as cur:
# ON CONFLICT DO NOTHING ensures idempotent writes
cur.execute("""
INSERT INTO payment_events (
event_id, payment_intent_id, amount_cents, currency,
customer_id, metadata, processed_at, event_timestamp
) VALUES (
%(event_id)s, %(payment_intent_id)s, %(amount_cents)s, %(currency)s,
%(customer_id)s, %(metadata)s::jsonb, %(processed_at)s,
to_timestamp(%(event_timestamp)s)
)
ON CONFLICT (event_id) DO NOTHING
""", record)
conn.commit()
log.info("warehouse.write_complete", event_id=record["event_id"], table=table,
rows_affected=cur.rowcount)
ON CONFLICT (event_id) DO NOTHING means if the same event is processed twice, the second insert is silently ignored. The event_id column needs a unique constraint or primary key definition in your schema. This is the database-level guarantee that ensures exactly-once semantics even when your application logic runs multiple times.
A webhook pipeline running silently is dangerous. You need to know: how many events are being received, how many are failing, how large the queue is growing, and how old the oldest unprocessed event is.
import structlog
from prometheus_client import Counter, Histogram, Gauge, start_http_server
# Metrics definitions
webhooks_received = Counter(
"webhooks_received_total",
"Total webhooks received",
["source", "event_type"]
)
webhooks_processed = Counter(
"webhooks_processed_total",
"Total webhooks processed",
["source", "event_type", "status"] # status: success, retry, dead_lettered
)
processing_duration = Histogram(
"webhook_processing_duration_seconds",
"Time to process a webhook event",
["source", "event_type"]
)
queue_depth = Gauge(
"webhook_queue_depth",
"Current number of events pending processing",
["source"]
)
def update_queue_depth_metric():
"""Call this periodically from a background thread or separate job."""
depth = redis_client.llen(WEBHOOK_QUEUE_KEY)
queue_depth.labels(source="stripe").set(depth)
The queue depth metric is particularly important. If it grows continuously, your workers aren't keeping up — you need to understand why before your backlog becomes unrecoverable. This kind of observability for data pipelines lets you set alerts like "queue depth > 1000 for more than 5 minutes" that fire before an ops problem becomes a data quality problem.
Beyond metrics, your structured logs should make it easy to trace an individual event's journey:
# Every log line for an event should include event_id, allowing you to filter
# the full processing history for any event in your log aggregation tool
log.info("webhook.received", event_id=event_id, event_type=event_type,
source="stripe", received_at=received_at)
log.info("webhook.processing_started", event_id=event_id,
worker_id=worker_id, attempt=envelope.retry_count + 1)
log.info("webhook.processing_complete", event_id=event_id,
duration_ms=duration_ms, rows_written=rows_written)
Your dead-letter queue accumulates events that couldn't be processed after all retries. This isn't a bin — it's a recovery mechanism. You need tooling to inspect and replay these events.
def inspect_dead_letter_queue(limit: int = 20) -> list[dict]:
"""Retrieve events from the dead-letter queue for inspection."""
raw_events = redis_client.lrange(DEAD_LETTER_KEY, 0, limit - 1)
return [json.loads(e) for e in raw_events]
def replay_dead_letter_event(event_id: str) -> bool:
"""
Find an event in the dead-letter queue by ID and re-queue it for processing.
Useful after fixing the bug that caused the original failure.
"""
raw_events = redis_client.lrange(DEAD_LETTER_KEY, 0, -1)
for i, raw in enumerate(raw_events):
event = json.loads(raw)
if event.get("event_id") == event_id:
# Reset retry count and re-queue
event["retry_count"] = 0
event.pop("failed_at", None)
event.pop("final_error", None)
redis_client.rpush(WEBHOOK_QUEUE_KEY, json.dumps(event))
# Remove from dead-letter queue
redis_client.lrem(DEAD_LETTER_KEY, 1, raw)
log.info("dead_letter.replayed", event_id=event_id)
return True
return False
def replay_all_dead_letters() -> int:
"""Replay all dead-letter events — use after fixing a widespread bug."""
events = redis_client.lrange(DEAD_LETTER_KEY, 0, -1)
replayed = 0
for raw in events:
event = json.loads(raw)
event["retry_count"] = 0
redis_client.rpush(WEBHOOK_QUEUE_KEY, json.dumps(event))
replayed += 1
redis_client.delete(DEAD_LETTER_KEY)
log.info("dead_letter.bulk_replay", count=replayed)
return replayed
Warning
Before replaying dead-letter events en masse, verify that your processing handlers are idempotent. Replaying events that write non-idempotently will create duplicate data. Test with a single event first, verify the result in your target system, then proceed with bulk replay.
Your webhook endpoint must be HTTPS. No exceptions. The signature validation we implemented provides cryptographic proof of message integrity, but without TLS the raw body is visible in transit — anyone on the network path could read the payload even if they can't forge a new valid signature.
The Redis list-based queue naturally supports multiple concurrent workers. Each BLPOP is atomic — Redis guarantees only one worker receives each message. You can run 10 workers reading from the same queue key without any coordination logic. When load increases, spin up more workers. When load decreases, let workers idle on the blocking pop.
For higher-volume scenarios, consider graduating to a dedicated queue system. AWS SQS with Lambda triggers, Google Cloud Pub/Sub, or Apache Kafka all provide stronger durability guarantees than Redis (which is in-memory by default and loses data on restart without AOF/RDB persistence configured).
Add rate limiting at the infrastructure level (your load balancer or API gateway) rather than in application code. A burst of legitimate webhooks from a large batch operation in your source system can look like a DDoS. Understand your source system's expected event rate and size your infrastructure accordingly. For context on throughput planning, data pipeline throughput estimation has useful frameworks for sizing decisions before you go to production.
Webhook payloads can change when source systems update their APIs. Adding a schema validation step in your receiver helps you catch breaking changes before they corrupt your pipeline. This pairs well with schema evolution strategies so you can handle Stripe adding new fields gracefully without crashing your handlers.
from pydantic import BaseModel, validator
from typing import Optional, Any
class StripeEventSchema(BaseModel):
id: str
type: str
created: int
data: dict[str, Any]
livemode: bool
@validator("type")
def validate_event_type(cls, v):
# Stripe event types follow "resource.action" pattern
if "." not in v:
raise ValueError(f"Unexpected event type format: {v}")
return v
# In your receiver, after JSON parsing:
try:
validated = StripeEventSchema(**payload)
except ValidationError as e:
log.warning("webhook.schema_validation_failed", errors=e.errors(),
event_id=payload.get("id"))
# Decision: reject (400) or accept and quarantine?
# Quarantine is usually safer — return 200, route to a review queue
Build a complete webhook ingestion pipeline for GitHub repository events. When a pull request is merged to main, your pipeline should capture the event and write a normalized record to a Postgres table for PR analytics.
Setup:
http://your-host/webhooks/github (use ngrok for local development)Build:
Receiver endpoint that validates GitHub's X-Hub-Signature-256 header (GitHub uses SHA-256, format: sha256=<hex>)
Schema validation for the pull request event payload, extracting:
pull_request.id (use as event_id)pull_request.numberpull_request.titlepull_request.merged_atpull_request.user.loginrepository.full_nameProcessing handler that only acts on action == "closed" events where pull_request.merged == True, and writes an idempotent record to a merged_pull_requests table
Dead-letter queue that captures events where the Postgres write fails
A replay script that reads events from the dead-letter queue and re-queues them
Validation checklist:
Parsing the body before signature validation
The signature is computed over the raw bytes. If you call request.json() first, the raw bytes are consumed and you can't verify the signature against them. Always read raw bytes first, then parse separately.
Not including the timestamp in the HMAC
Some implementations verify the signature but skip the timestamp check. This means a captured valid request can be replayed indefinitely. Always validate that the timestamp is within tolerance.
Using Redis without persistence
Redis is in-memory by default. If your Redis instance restarts, your queue is gone. Enable AOF (append-only file) persistence for the instance backing your webhook queue, or use a persistence-first queue like SQS or Postgres-backed pgmq.
Returning 500 when processing fails instead of accepting and queuing
If you return 5xx from your endpoint, the sender will retry. This is actually the right behavior when the queue is unavailable — you want the sender to retry. But if you return 5xx when your processing fails (the queue write succeeded, processing failed later), you'll cause the sender to resend an event you've already persisted, defeating your deduplication.
Queue growing unbounded without monitoring
If your workers crash and stay down, events accumulate in the queue. Without queue depth alerting, you may not notice for hours. Set an alert for queue depth exceeding your expected 15-minute backlog at peak throughput.
Forgetting to handle unhandled event types gracefully
If your router throws an exception on unknown event types, valid events of unexpected types will retry until they hit the dead-letter queue. Log and skip unknown types; don't treat them as errors.
Tip
When debugging signature failures during local development, print the raw header value and the raw body bytes together. A common source of mismatch is middleware that modifies the request body (gzip decompression, body parsing middleware) before your validation code runs. Validate signatures as the very first thing your handler does.
You've built a production-grade webhook ingestion layer with:
The patterns here scale from a single Stripe integration to dozens of concurrent webhook sources. The key is keeping the receiver layer simple and fast — it's a mailbox, not a processing engine — and investing in the worker layer where the real complexity lives.
From here, consider exploring understanding data pipeline triggers to see how webhook-driven ingestion compares to time-based scheduling and sensor patterns in a unified orchestration model. If you're routing events from multiple webhook sources into different downstream tables, the fan-out pipeline design pattern is a natural next topic. And if your webhook volume is high enough that you're starting to think about throughput limits, backpressure and bottleneck diagnosis will give you the tools to find and fix the constraints.