🪝 Webhooks

Process webhooks with signature verification, idempotency, and custom handlers.

🎯 Overview

Function	Purpose
verify_webhook_signature	HMAC-SHA256 signature verification
check_idempotency	Prevent duplicate processing
log_webhook_event	Persist event to database
process_webhook	Main orchestration function
handle_webhook_request	FastHTML route handler

---

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    Webhook Processing Flow                      │
├─────────────────────────────────────────────────────────────────┤
│  Request → Verify Signature → Check Idempotency → Log Event    │
│     ↓           ✅              ✅ (not duplicate)              │
│  Execute Handler → Update Status → Return Response              │
└─────────────────────────────────────────────────────────────────┘

🔐 Signature Verification

Function	Purpose
verify_webhook_signature	HMAC-SHA256 with timing-safe comparison

from nbdev.showdoc import *

---

source

verify_webhook_signature

def verify_webhook_signature(
    payload:str, # Raw request body as string
    signature:str, # Signature from header (format: "sha256=<hex>")
    secret:Optional=None, # Secret key, defaults to WEBHOOK_SECRET env var
)->bool:

Verify HMAC-SHA256 signature for webhook payload. Returns True if valid.

🔄 Idempotency

Function	Purpose
check_idempotency	Check if event already processed

---

source

check_idempotency

def check_idempotency(
    db:Database, # Tenant database connection
    idempotency_key:str, # Unique key for this webhook event
)->bool:

Check if webhook event already processed. Returns True if duplicate.

📝 Event Logging

Function	Purpose
log_webhook_event	Persist event to database
update_webhook_status	Update status after processing

---

source

log_webhook_event

def log_webhook_event(
    db:Database, # Tenant database connection
    webhook_id:str, # Unique webhook ID
    source:str, # Source system (e.g., "stripe", "github")
    event_type:str, # Event type (e.g., "payment.success")
    payload:Dict, # Full webhook payload
    signature:str, # Request signature
    idempotency_key:str, # Idempotency key
    status:str='pending', # Status: pending, processing, completed, failed
):

Log webhook event to database

Update Webhook Status

---

source

update_webhook_status

def update_webhook_status(
    db:Database, # Tenant database connection
    webhook_id:str, # Webhook ID to update
    status:str, # New status
    error_message:Optional=None, # Optional error message
):

Update webhook event status and processed timestamp

⚡ Process Webhook

Function	Purpose
process_webhook	Main orchestration function
handle_webhook_request	FastHTML route handler

---

source

process_webhook

def process_webhook(
    db:Database, # Tenant database connection
    webhook_id:str, # Unique webhook ID
    source:str, # Source system
    event_type:str, # Event type
    payload:Dict, # Webhook payload
    signature:str, # Request signature
    idempotency_key:str, # Idempotency key
    raw_body:str, # Raw request body for signature verification
    handler:Callable, # App-specific webhook handler function
    secret:Optional=None, # Optional webhook secret
)->Dict:

Process webhook with verification, idempotency, and custom handler execution

from nbdev.showdoc import show_doc

---

source

verify_webhook_signature

def verify_webhook_signature(
    payload:str, # Raw request body as string
    signature:str, # Signature from header (format: "sha256=<hex>")
    secret:Optional=None, # Secret key, defaults to WEBHOOK_SECRET env var
)->bool:

Verify HMAC-SHA256 signature for webhook payload. Returns True if valid.

---

source

check_idempotency

def check_idempotency(
    db:Database, # Tenant database connection
    idempotency_key:str, # Unique key for this webhook event
)->bool:

Check if webhook event already processed. Returns True if duplicate.

---

source

log_webhook_event

def log_webhook_event(
    db:Database, # Tenant database connection
    webhook_id:str, # Unique webhook ID
    source:str, # Source system (e.g., "stripe", "github")
    event_type:str, # Event type (e.g., "payment.success")
    payload:Dict, # Full webhook payload
    signature:str, # Request signature
    idempotency_key:str, # Idempotency key
    status:str='pending', # Status: pending, processing, completed, failed
):

Log webhook event to database

---

source

update_webhook_status

def update_webhook_status(
    db:Database, # Tenant database connection
    webhook_id:str, # Webhook ID to update
    status:str, # New status
    error_message:Optional=None, # Optional error message
):

Update webhook event status and processed timestamp

---

source

process_webhook

def process_webhook(
    db:Database, # Tenant database connection
    webhook_id:str, # Unique webhook ID
    source:str, # Source system
    event_type:str, # Event type
    payload:Dict, # Webhook payload
    signature:str, # Request signature
    idempotency_key:str, # Idempotency key
    raw_body:str, # Raw request body for signature verification
    handler:Callable, # App-specific webhook handler function
    secret:Optional=None, # Optional webhook secret
)->Dict:

Process webhook with verification, idempotency, and custom handler execution

---

source

handle_webhook_request

def handle_webhook_request(
    request, # FastHTML request object
    db:Database, # Tenant database instance
    source:str, # Webhook source identifier
    handler:Callable, # App-specific handler function
    signature_header:str='X-Webhook-Signature', # Header containing signature
    idempotency_header:str='X-Idempotency-Key', # Header containing idempotency key
    event_type_field:str='type', # Field in payload containing event type
    secret:Optional=None, # Optional webhook secret
)->tuple: # Returns (response_dict, status_code)

FastHTML route handler for webhook requests.

Example: @app.post(‘/webhooks/stripe’) async def stripe_webhook(request): return await handle_webhook_request( request=request, db=get_tenant_db(request), source=‘stripe’, handler=handle_stripe_event, signature_header=‘X-Stripe-Signature’, run_in_background=True )