How to Handle Webhooks in Django

Creating a Webhook View

Django's CSRF protection will block webhook POST requests by default. You need to exempt your webhook views from CSRF checking using the @csrf_exempt decorator:

from django.http import JsonResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
import json

@csrf_exempt @require_POST def webhook_handler(request): try: payload = json.loads(request.body) except json.JSONDecodeError: return HttpResponseBadRequest('Invalid JSON')

event_type = payload.get('type') print(f'Webhook received: {event_type}')

if event_type == 'payment.completed': handle_payment(payload['data']) elif event_type == 'user.created': handle_user_created(payload['data'])

return JsonResponse({'received': True}) ```

The @require_POST decorator rejects non-POST requests. The request.body attribute contains the raw bytes, which you can parse with json.loads(). Always keep the raw body available for signature verification.

Signature Verification

Django gives you the raw body via request.body, which makes signature verification straightforward:

import hmac
import hashlib

@csrf_exempt @require_POST def stripe_webhook(request): payload = request.body sig_header = request.META.get('HTTP_STRIPE_SIGNATURE', '') secret = settings.STRIPE_WEBHOOK_SECRET

# Parse Stripe signature elements = dict( item.split('=', 1) for item in sig_header.split(',') )

signed_payload = f"{elements['t']}.{payload.decode('utf-8')}" expected = hmac.new( secret.encode('utf-8'), signed_payload.encode('utf-8'), hashlib.sha256 ).hexdigest()

if not hmac.compare_digest(expected, elements['v1']): return HttpResponseBadRequest('Invalid signature')

event = json.loads(payload) process_stripe_event(event)

return JsonResponse({'received': True}) ```

Note that Django stores HTTP headers in request.META with an HTTP_ prefix and uppercase names with hyphens replaced by underscores. So Stripe-Signature becomes HTTP_STRIPE_SIGNATURE. Use hmac.compare_digest() for timing-safe comparison.

URL Configuration

Add your webhook endpoint to your URL configuration:

# urls.py
from django.urls import path
from . import views

urlpatterns = [ path('api/webhooks/stripe/', views.stripe_webhook, name='stripe-webhook'), path('api/webhooks/github/', views.github_webhook, name='github-webhook'), path('api/webhooks/', views.generic_webhook, name='generic-webhook'), ] ```

For Django REST Framework, you can use a viewset approach instead:

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import AllowAny
from rest_framework.response import Response

@api_view(['POST']) @permission_classes([AllowAny]) def webhook_handler(request): # DRF handles parsing automatically # Access raw body with request.body (before DRF parses it) event = request.data process_event(event) return Response({'received': True}) ```

Make sure your webhook URLs are not behind Django's authentication middleware. Webhook endpoints should be publicly accessible — verification is handled by signature checking, not by Django's auth system.

Async Processing with Celery

For long-running webhook processing, use Celery to handle events asynchronously:

from celery import shared_task

@shared_task def process_webhook_event(event_type, event_data): if event_type == 'payment.completed': order = Order.objects.get(payment_id=event_data['id']) order.status = 'paid' order.save() send_confirmation_email(order) elif event_type == 'subscription.canceled': subscription = Subscription.objects.get( external_id=event_data['id'] ) subscription.deactivate()

@csrf_exempt @require_POST def webhook_handler(request): event = json.loads(request.body)

# Idempotency check event_id = event['id'] if ProcessedEvent.objects.filter(event_id=event_id).exists(): return JsonResponse({'received': True}) # Skip duplicate

ProcessedEvent.objects.create(event_id=event_id)

# Queue for async processing process_webhook_event.delay(event['type'], event['data'])

return JsonResponse({'received': True}) ```

This pattern responds to the webhook immediately (keeping the provider happy) and processes the event in a Celery worker. The idempotency check prevents duplicate processing.

Testing with ReqPour

During development, use ReqPour to relay webhooks to your Django dev server:

python manage.py runserver 0.0.0.0:8000
# In another terminal:
npx reqpour relay --to http://localhost:8000

For Django's test framework, create test cases using captured webhook payloads:

from django.test import TestCase, Client

class WebhookTests(TestCase): def test_payment_completed(self): client = Client() payload = { 'id': 'evt_test_123', 'type': 'payment.completed', 'data': {'id': 'pay_456', 'amount': 2500} } response = client.post( '/api/webhooks/', data=json.dumps(payload), content_type='application/json' ) self.assertEqual(response.status_code, 200) # Verify order was updated order = Order.objects.get(payment_id='pay_456') self.assertEqual(order.status, 'paid') ```

Capture real webhook payloads from the ReqPour dashboard and use them as fixtures in your Django tests for realistic test data.