Advanced Webhook Handling
Beyond basic webhook setup, here are advanced patterns and best practices.
Webhook Signature Verification
Always verify webhook signatures to ensure requests are from Paystack:
from djpaystack.webhooks.handlers import verify_webhook_signature
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
import json
@require_http_methods(["POST"])
def webhook_with_verification(request):
"""Webhook with signature verification"""
# Get the webhook secret from settings
from django.conf import settings
webhook_secret = settings.PAYSTACK.get('WEBHOOK_SECRET')
# Verify signature
if not verify_webhook_signature(request, webhook_secret):
return JsonResponse({'error': 'Invalid signature'}, status=401)
# Process the webhook
try:
payload = json.loads(request.body)
event = payload.get('event')
if event == 'charge.success':
handle_charge_success(payload['data'])
elif event == 'charge.failed':
handle_charge_failed(payload['data'])
return JsonResponse({'status': 'ok'})
except Exception as e:
return JsonResponse({'error': str(e)}, status=500)
Idempotent Processing
Process webhooks idempotently so duplicate events don’t cause issues:
from django.db import models
from django.db.models import Q
import uuid
class WebhookEvent(models.Model):
"""Track processed webhook events"""
event_id = models.CharField(max_length=255, unique=True)
event = models.CharField(max_length=100)
data = models.JSONField()
processed = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
def process_webhook(payload):
"""Process webhook idempotently"""
event_id = payload.get('id')
# Check if already processed
try:
webhook_event = WebhookEvent.objects.get(event_id=event_id)
if webhook_event.processed:
return # Already processed
except WebhookEvent.DoesNotExist:
webhook_event = WebhookEvent.objects.create(
event_id=event_id,
event=payload.get('event'),
data=payload.get('data', {})
)
# Process the event
try:
handle_event(payload)
webhook_event.processed = True
webhook_event.save()
except Exception as e:
# Retry on next attempt
raise
Async Webhook Processing
Use Celery for async webhook processing to avoid timeouts:
from celery import shared_task
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
import json
@shared_task
def process_webhook_async(event, data):
"""Process webhook asynchronously"""
if event == 'charge.success':
handle_charge_success(data)
elif event == 'charge.failed':
handle_charge_failed(data)
# ... other event handlers
@require_http_methods(["POST"])
def webhook_async(request):
"""Async webhook handler"""
payload = json.loads(request.body)
# Queue for async processing
process_webhook_async.delay(
payload.get('event'),
payload.get('data')
)
# Return immediately
return JsonResponse({'status': 'received'})
Webhook Retry Logic
Implement retry logic for failed webhook processing:
from django.db import models
from django.utils import timezone
from django.core.exceptions import ValidationError
import json
class WebhookLog(models.Model):
"""Log webhook attempts"""
STATUS_CHOICES = [
('pending', 'Pending'),
('success', 'Success'),
('failed', 'Failed'),
]
event_id = models.CharField(max_length=255)
event = models.CharField(max_length=100)
data = models.JSONField()
status = models.CharField(max_length=20, choices=STATUS_CHOICES)
error = models.TextField(blank=True)
attempts = models.IntegerField(default=0)
last_attempt = models.DateTimeField(null=True)
created_at = models.DateTimeField(auto_now_add=True)
max_retries = 3
retry_delay = 300 # 5 minutes
def process_webhook_with_retry(payload):
"""Process webhook with automatic retry"""
event_id = payload.get('id')
log, created = WebhookLog.objects.get_or_create(
event_id=event_id,
defaults={
'event': payload.get('event'),
'data': payload.get('data', {}),
'status': 'pending'
}
)
if log.attempts >= log.max_retries:
log.status = 'failed'
log.save()
return
try:
handle_event(payload)
log.status = 'success'
except Exception as e:
log.status = 'pending'
log.error = str(e)
log.attempts += 1
log.last_attempt = timezone.now()
# Could schedule retry with Celery
log.save()
Webhook Monitoring
Monitor webhook health and failures:
from django.core.management.base import BaseCommand
from django.db.models import Q
from django.utils import timezone
from datetime import timedelta
from .models import WebhookLog
class Command(BaseCommand):
"""Check webhook health"""
def handle(self, *args, **options):
now = timezone.now()
five_minutes_ago = now - timedelta(minutes=5)
# Find failures
failures = WebhookLog.objects.filter(
status='failed',
created_at__gte=five_minutes_ago
)
if failures.exists():
print(f"⚠️ {failures.count()} webhook failures in last 5 minutes")
for log in failures:
print(f" - {log.event}: {log.error}")
# Find stuck pending
stuck = WebhookLog.objects.filter(
status='pending',
attempts__gte=3,
last_attempt__lt=five_minutes_ago
)
if stuck.exists():
print(f"❌ {stuck.count()} stuck webhooks")
Best Practices Summary
Always verify signatures - Never process unverified webhooks
Respond quickly - Return 200 within 3 seconds
Process idempotently - Handle duplicate events gracefully
Use async processing - Don’t block webhook responses
Implement retry logic - Handle temporary failures
Log everything - Keep audit trail of webhook events
Monitor health - Track failure rates
Test thoroughly - Use ngrok for local testing
Testing Webhook Security
Test webhook signature verification:
from django.test import TestCase, Client
from django.conf import settings
import hmac
import hashlib
import json
class WebhookSecurityTestCase(TestCase):
def test_invalid_signature_rejected(self):
"""Test that invalid signatures are rejected"""
client = Client()
payload = {
'event': 'charge.success',
'data': {'reference': 'test'}
}
# Send with invalid signature
response = client.post(
'/api/webhooks/paystack/',
data=json.dumps(payload),
content_type='application/json',
HTTP_X_PAYSTACK_SIGNATURE='invalid_signature'
)
self.assertEqual(response.status_code, 401)
def test_valid_signature_accepted(self):
"""Test that valid signatures are accepted"""
client = Client()
payload = {'event': 'charge.success', 'data': {}}
webhook_secret = settings.PAYSTACK['WEBHOOK_SECRET']
# Generate valid signature
signature = hmac.new(
webhook_secret.encode(),
json.dumps(payload).encode(),
hashlib.sha512
).hexdigest()
response = client.post(
'/api/webhooks/paystack/',
data=json.dumps(payload),
content_type='application/json',
HTTP_X_PAYSTACK_SIGNATURE=signature
)
self.assertEqual(response.status_code, 200)