Webhook with Django — Complete Guide
In this tutorial, you will learn about Webhook with Django. We cover key concepts, practical examples, and best practices to help you master this topic.
Learn to handle Webhooks in Django: CSRF exemption, signature verification, raw body access, async task processing with Celery, logging, and best practices for Django webhook endpoints.
What You Learn
You will learn how to implement webhook endpoints in Django: bypass CSRF for external POST requests, access raw request bodies for HMAC verification, use Django views for webhook handling, queue processing with Celery, and structure webhook code in Django projects.
Why It Matters
Django's built-in CSRF protection blocks external POST requests by default. Without proper configuration, webhook providers cannot reach your Django endpoints. Understanding Django-specific webhook patterns ensures your endpoints are both secure and accessible.
Real-World Use
DodaTech's Django-based billing service handles webhooks from Stripe and PayPal. The endpoint is CSRF-exempt, verifies signatures, queues payment processing tasks to Celery, and logs all webhook activity to Django admin for audit.
Basic Webhook View
import hashlib
import hmac
import json
from django.http import JsonResponse, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
WEBHOOK_SECRET = b'whsec_your_secret_here'
@csrf_exempt
@require_POST
def webhook_receiver(request):
# Get raw body for signature verification
raw_body = request.body
# Verify signature
signature = request.headers.get('X-Webhook-Signature', '')
if not verify_signature(raw_body, signature, WEBHOOK_SECRET):
return HttpResponse('Invalid signature', status=401)
# Verify timestamp
timestamp = request.headers.get('X-Webhook-Timestamp', '')
if is_stale(timestamp):
return HttpResponse('Stale webhook', status=400)
# Parse payload
try:
payload = json.loads(raw_body)
except json.JSONDecodeError:
return HttpResponse('Invalid JSON', status=400)
# Acknowledge immediately
# Process asynchronously
process_webhook.delay(payload)
return JsonResponse({'status': 'accepted'})
def verify_signature(raw_body, signature_header, secret):
if not signature_header or not raw_body:
return False
expected_sig = signature_header.replace('sha256=', '')
computed_sig = hmac.new(
secret,
raw_body,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed_sig, expected_sig)
Expected output: The view is CSRF-exempt (required for external POST), verifies the HMAC signature, checks timestamp, parses JSON, and queues async processing. Invalid requests return appropriate HTTP status codes.
Celery Task Processing
# tasks.py
from celery import shared_task
from django.utils import timezone
from .models import WebhookLog, Payment
@shared_task(bind=True, max_retries=3)
def process_webhook(self, payload):
event_type = payload.get('type') or payload.get('event')
data = payload.get('data', {})
log_entry = WebhookLog.objects.create(
event_type=event_type,
payload=payload,
status='processing',
)
try:
if event_type == 'payment.succeeded':
handle_payment_succeeded(data)
elif event_type == 'payment.failed':
handle_payment_failed(data)
elif event_type == 'customer.subscription.updated':
handle_subscription_updated(data)
else:
logger.warning(f'Unknown event type: {event_type}')
log_entry.status = 'completed'
log_entry.save()
except Exception as exc:
log_entry.status = 'failed'
log_entry.error_message = str(exc)
log_entry.save()
raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)
def handle_payment_succeeded(data):
payment_id = data.get('id')
amount = data.get('amount', 0)
customer_id = data.get('customer')
Payment.objects.create(
provider_payment_id=payment_id,
amount=amount / 100,
customer_id=customer_id,
status='succeeded',
received_at=timezone.now(),
)
# Update order status
order = Order.objects.get(payment_intent_id=payment_id)
order.status = 'paid'
order.save()
Expected output: Celery task processes the webhook asynchronously. Retries up to 3 times with exponential backoff on failure. Webhook log tracks processing status and errors.
Django Admin Integration
# admin.py
from django.contrib import admin
from .models import WebhookLog, WebhookEndpoint
@admin.register(WebhookLog)
class WebhookLogAdmin(admin.ModelAdmin):
list_display = [
'event_type', 'status', 'created_at', 'duration_ms'
]
list_filter = ['status', 'event_type', 'created_at']
search_fields = ['event_type', 'error_message']
readonly_fields = [
'event_type', 'payload', 'status',
'error_message', 'created_at', 'completed_at'
]
def duration_ms(self, obj):
if obj.completed_at and obj.created_at:
delta = obj.completed_at - obj.created_at
return int(delta.total_seconds() * 1000)
return None
duration_ms.short_description = 'Duration (ms)'
@admin.register(WebhookEndpoint)
class WebhookEndpointAdmin(admin.ModelAdmin):
list_display = [
'name', 'url', 'is_active', 'events_count', 'last_delivery'
]
fields = [
'name', 'url', 'secret', 'events',
'is_active', 'created_at'
]
readonly_fields = ['created_at']
Expected output: Django admin provides visibility into webhook processing. Operators can inspect failed webhooks, retry processing, and manage endpoint configurations.
Provider-Specific Views
# views.py - Provider-specific webhook handlers
import stripe
from django.conf import settings
from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
@require_POST
def stripe_webhook(request):
"""Stripe-specific webhook handler."""
raw_body = request.body
sig_header = request.headers.get('Stripe-Signature')
try:
event = stripe.Webhook.construct_event(
raw_body,
sig_header,
settings.STRIPE_WEBHOOK_SECRET
)
except ValueError:
return HttpResponse('Invalid payload', status=400)
except stripe.error.SignatureVerificationError:
return HttpResponse('Invalid signature', status=401)
# Map Stripe events to handlers
event_map = {
'payment_intent.succeeded': handle_stripe_payment_success,
'payment_intent.payment_failed': handle_stripe_payment_failed,
'customer.subscription.updated': handle_stripe_subscription,
'charge.refunded': handle_stripe_refund,
}
handler = event_map.get(event.type)
if handler:
handler(event.data.object)
return JsonResponse({'status': 'success'})
@csrf_exempt
@require_POST
def github_webhook(request):
"""GitHub-specific webhook handler."""
raw_body = request.body
signature = request.headers.get('X-Hub-Signature-256', '')
event_type = request.headers.get('X-GitHub-Event', '')
if not verify_signature(raw_body, signature, settings.GITHUB_WEBHOOK_SECRET):
return HttpResponse('Invalid signature', status=401)
payload = json.loads(raw_body)
event_map = {
'push': handle_github_push,
'pull_request': handle_github_pr,
'issues': handle_github_issue,
'release': handle_github_release,
}
handler = event_map.get(event_type)
if handler:
handler(payload)
return JsonResponse({'status': 'success'})
Expected output: Each provider gets its own view with provider-specific verification logic. Stripe uses stripe.Webhook.construct_event. GitHub uses custom HMAC verification with the X-GitHub-Event header.
Webhook URL Configuration
# urls.py
from django.urls import path
from . import views
urlpatterns = [
# Generic webhook endpoint
path('webhook/', views.webhook_receiver, name='webhook-receiver'),
# Provider-specific endpoints
path('webhooks/stripe/', views.stripe_webhook, name='stripe-webhook'),
path('webhooks/github/', views.github_webhook, name='github-webhook'),
path('webhooks/sendgrid/', views.sendgrid_webhook, name='sendgrid-webhook'),
# Dashboard
path('webhooks/logs/', views.webhook_logs, name='webhook-logs'),
path('webhooks/retry/<int:log_id>/', views.retry_webhook, name='retry-webhook'),
]
Expected output: URL configuration maps each provider to its own view. Generic webhook endpoint for custom providers. Dashboard endpoints for monitoring and manual retry.
Common Mistakes
1. Forgetting CSRF Exemption
Django's CSRF middleware blocks all POST requests from external origins. Webhook endpoints must be decorated with @csrf_exempt. Without it, providers receive 403 Forbidden responses.
2. Using request.POST Instead of request.body
Django's request.POST only contains form-encoded data. Webhook payloads are typically JSON in the request body. Use request.body to access the raw bytes. Parse with json.loads().
3. Processing Synchronously in the View
Database writes and external API calls in the view cause timeouts. Use Celery tasks for processing. The view should only verify, acknowledge (200), and queue.
4. Not Logging Webhook Activity
Without logging, debugging webhook issues in Django admin is impossible. Log every webhook: provider, event type, payload, verification result, processing status. Use Django models for the log.
5. Missing URL Configuration
Webhook views need proper URL routing. Ensure URLs are accessible from the internet. Use ngrok for local testing. Verify the full URL path matches the provider configuration.
Practice Questions
1. Why do Django webhook views need @csrf_exempt?
Django's CSRF middleware requires a CSRF token for all POST requests from browsers. Webhook providers do not include CSRF tokens. @csrf_exempt disables CSRF protection for the specific view.
2. How do you access the raw request body in Django?
Use request.body which returns bytes. Do not use request.POST or request.data (DRF) as they parse the body. Raw bytes are needed for HMAC signature verification.
3. How do you Process webhooks asynchronously in Django?
Use Celery tasks. Create a shared_task decorated function. Call it with .delay() from the view. The task runs in a worker process, freeing the web server to handle more requests.
4. How do you retry failed webhook processing in Django?
Celery tasks support automatic retry with the bind=True and max_retries parameters. Use self.retry() with exponential backoff. Log retry attempts in the database for visibility.
Challenge
Build a Django webhook system: CSRF-exempt endpoints for Stripe, GitHub, and custom providers, Celery task processing with retries, webhook log model with Django admin integration, retry button in admin, and provider-specific signature verification for each endpoint.
FAQ
Mini Project: Django Webhook Manager
Build a Django app that: manages webhook endpoint configurations in admin, receives webhooks from Stripe and GitHub, verifies provider-specific signatures, processes events with Celery tasks, logs all activity, provides a webhook log viewer in admin, and supports manual retry of failed webhooks.
What's Next
Now that you can handle webhooks in Django, learn Webhook with FastAPI for async Python webhook consumers.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro