Hook Mesh
Log inStart Free
Back to Blog
Hook Mesh Team

Building Webhooks with Django and Hook Mesh

A complete tutorial for Django developers on implementing webhooks—sending events from views and signals, receiving webhooks securely, signature verification, and integrating with Hook Mesh for reliable delivery.

Building Webhooks with Django and Hook Mesh

Building Webhooks with Django and Hook Mesh

Django powers thousands of SaaS apps. This tutorial covers sending webhooks when events occur and receiving webhooks from external services securely. For more Python-specific patterns, see sending webhooks with Python and receiving and verifying webhooks in Python.

Understanding Webhooks in Django

Webhooks are HTTP callbacks—when something happens, send a POST request to a subscriber's URL. Django lets you trigger them from views or use the signal system to decouple webhook logic. The challenge is handling failures, implementing retries, and managing multiple subscribers. See our webhook implementation guides for patterns across languages.

Sending Webhooks from Django Views

# views.py
import json
import requests
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from django.contrib.auth.decorators import login_required
from .models import Order

@login_required
@require_POST
def complete_order(request, order_id):
    order = Order.objects.get(id=order_id, user=request.user)
    order.status = 'completed'
    order.save()

    # Send webhook notification
    webhook_payload = {
        'event': 'order.completed',
        'data': {
            'order_id': str(order.id),
            'user_id': str(order.user.id),
            'total': str(order.total),
            'completed_at': order.updated_at.isoformat()
        }
    }

    # Direct webhook call (not recommended for production)
    try:
        requests.post(
            'https://subscriber.example.com/webhooks',
            json=webhook_payload,
            timeout=5
        )
    except requests.RequestException:
        pass  # Webhook delivery failed silently

    return JsonResponse({'status': 'completed'})

This has limitations: blocking requests, silent failures, no retries. For production, use Django signals.

Using Django Signals for Webhook Events

Signals decouple webhook sending from business logic—trigger actions on events without modifying original code:

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order, Subscription

@receiver(post_save, sender=Order)
def order_saved_handler(sender, instance, created, **kwargs):
    if instance.status == 'completed':
        send_webhook_event('order.completed', {
            'order_id': str(instance.id),
            'user_id': str(instance.user.id),
            'total': str(instance.total),
            'completed_at': instance.updated_at.isoformat()
        })

@receiver(post_save, sender=Subscription)
def subscription_saved_handler(sender, instance, created, **kwargs):
    if created:
        event_type = 'subscription.created'
    else:
        event_type = 'subscription.updated'

    send_webhook_event(event_type, {
        'subscription_id': str(instance.id),
        'user_id': str(instance.user.id),
        'plan': instance.plan.name,
        'status': instance.status
    })

Don't forget to import your signals in your app's apps.py:

# apps.py
from django.apps import AppConfig

class CoreConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'core'

    def ready(self):
        import core.signals  # noqa

Integrating Hook Mesh for Reliable Delivery

Let Hook Mesh handle delivery complexity. The Python SDK makes integration straightforward:

# webhook_service.py
from hookmesh import HookMesh
from django.conf import settings

hookmesh = HookMesh(settings.HOOKMESH_API_KEY)

def send_webhook_event(event_type: str, data: dict, customer_id: str = None):
    """Send a webhook event through Hook Mesh."""
    hookmesh.events.send(
        event_type=event_type,
        data=data,
        customer_id=customer_id  # Optional: for multi-tenant apps
    )

Update your signals to use this service:

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from .webhook_service import send_webhook_event

@receiver(post_save, sender=Order)
def order_saved_handler(sender, instance, created, **kwargs):
    if instance.status == 'completed':
        send_webhook_event(
            'order.completed',
            {
                'order_id': str(instance.id),
                'user_id': str(instance.user.id),
                'total': str(instance.total)
            },
            customer_id=str(instance.user.organization_id)
        )

Hook Mesh handles retries with exponential backoff, maintains delivery logs, and provides customer dashboards.

Async Webhook Sending with Celery

Celery pairs perfectly with Django for async webhook sending in high-throughput apps:

# tasks.py
from celery import shared_task
from .webhook_service import send_webhook_event

@shared_task(bind=True, max_retries=3)
def send_webhook_async(self, event_type: str, data: dict, customer_id: str = None):
    try:
        send_webhook_event(event_type, data, customer_id)
    except Exception as exc:
        self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

Then in your signals:

from .tasks import send_webhook_async

@receiver(post_save, sender=Order)
def order_saved_handler(sender, instance, created, **kwargs):
    if instance.status == 'completed':
        send_webhook_async.delay(
            'order.completed',
            {'order_id': str(instance.id), 'total': str(instance.total)},
            str(instance.user.organization_id)
        )

Django 4.1+ supports async views with httpx, but Celery is recommended for production.

Receiving Webhooks in Django

URL Configuration

Start by setting up a dedicated URL for incoming webhooks:

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

urlpatterns = [
    path('webhooks/stripe/', views.stripe_webhook, name='stripe_webhook'),
    path('webhooks/github/', views.github_webhook, name='github_webhook'),
]

CSRF Exemption for Webhook Endpoints

Exempt webhook endpoints from Django's CSRF protection—external services don't have your CSRF token:

# views.py
import json
from django.http import JsonResponse, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

@csrf_exempt
@require_POST
def stripe_webhook(request):
    payload = request.body  # Raw bytes - important for signature verification
    signature = request.headers.get('Stripe-Signature', '')

    # Verify and process the webhook
    # ...

    return HttpResponse(status=200)

The @csrf_exempt decorator is critical. Without it, Django returns a 403 Forbidden response to all webhook requests.

Accessing Raw Request Body

Access the raw request body exactly as received via request.body:

@csrf_exempt
@require_POST
def incoming_webhook(request):
    # request.body returns bytes - the raw, unmodified payload
    raw_payload = request.body

    # For string operations, decode explicitly
    payload_string = raw_payload.decode('utf-8')

    # Parse JSON after verification
    try:
        data = json.loads(payload_string)
    except json.JSONDecodeError:
        return JsonResponse({'error': 'Invalid JSON'}, status=400)

    return JsonResponse({'received': True})

Never use request.POST or call json.loads() before signature verification—these can alter or reparse the data, causing signature mismatches.

Webhook Signature Verification

Signature verification ensures webhooks come from the claimed source. Most providers use HMAC-SHA256. See webhook security best practices for more authentication options:

# webhook_verification.py
import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify HMAC-SHA256 webhook signature."""
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    # Use compare_digest to prevent timing attacks
    return hmac.compare_digest(signature, expected_signature)

Now use it in your view:

# views.py
import json
import os
from django.http import JsonResponse, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from .webhook_verification import verify_webhook_signature

@csrf_exempt
@require_POST
def payment_webhook(request):
    payload = request.body
    signature = request.headers.get('X-Webhook-Signature', '')
    secret = os.environ.get('WEBHOOK_SECRET')

    if not verify_webhook_signature(payload, signature, secret):
        return JsonResponse({'error': 'Invalid signature'}, status=401)

    # Signature valid - safe to process
    data = json.loads(payload)
    event_type = data.get('event')

    if event_type == 'payment.completed':
        handle_payment_completed(data['data'])
    elif event_type == 'payment.failed':
        handle_payment_failed(data['data'])

    return HttpResponse(status=200)

def handle_payment_completed(payment_data):
    # Process successful payment
    pass

def handle_payment_failed(payment_data):
    # Handle failed payment
    pass

Class-Based Views for Webhooks

For complex webhook handling, class-based views provide better organization:

# views.py
import json
from django.http import JsonResponse, HttpResponse
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
from .webhook_verification import verify_webhook_signature

@method_decorator(csrf_exempt, name='dispatch')
class WebhookView(View):
    webhook_secret_env = 'WEBHOOK_SECRET'

    def get_webhook_secret(self):
        return os.environ.get(self.webhook_secret_env)

    def verify_signature(self, request):
        signature = request.headers.get('X-Webhook-Signature', '')
        return verify_webhook_signature(
            request.body,
            signature,
            self.get_webhook_secret()
        )

    def post(self, request):
        if not self.verify_signature(request):
            return JsonResponse({'error': 'Invalid signature'}, status=401)

        data = json.loads(request.body)
        event_type = data.get('event')
        handler = getattr(self, f'handle_{event_type.replace(".", "_")}', None)

        if handler:
            handler(data['data'])

        return HttpResponse(status=200)

    def handle_order_completed(self, data):
        # Process order.completed event
        pass

Testing Your Webhook Endpoints

Use Django's test client to verify webhook implementations:

# tests.py
import json
import hmac
import hashlib
from django.test import TestCase, Client

class WebhookTests(TestCase):
    def setUp(self):
        self.client = Client()
        self.secret = 'test_secret'

    def sign_payload(self, payload: bytes) -> str:
        return hmac.new(
            self.secret.encode(),
            payload,
            hashlib.sha256
        ).hexdigest()

    def test_valid_webhook(self):
        payload = json.dumps({'event': 'order.completed', 'data': {}}).encode()
        signature = self.sign_payload(payload)

        response = self.client.post(
            '/webhooks/orders/',
            data=payload,
            content_type='application/json',
            HTTP_X_WEBHOOK_SIGNATURE=signature
        )

        self.assertEqual(response.status_code, 200)

    def test_invalid_signature_rejected(self):
        payload = json.dumps({'event': 'order.completed', 'data': {}}).encode()

        response = self.client.post(
            '/webhooks/orders/',
            data=payload,
            content_type='application/json',
            HTTP_X_WEBHOOK_SIGNATURE='invalid_signature'
        )

        self.assertEqual(response.status_code, 401)

Production Considerations

Respond quickly: Return 200 immediately, queue heavy processing with Celery.

Idempotency: Store processed event IDs to detect duplicates. See webhook idempotency guide for patterns.

Logging: Log all incoming webhooks with signatures and verification results.

Timeouts: External services expect 5-30 second responses. Long-running handlers cause failures.

Conclusion

Django provides signals for clean architecture, views for secure receiving, and Celery for async processing. Hook Mesh removes the sending infrastructure burden—retries, subscriber management, and dashboards. Get started with Hook Mesh.

Related Posts

Receive and Verify Webhooks in Python: A Complete Guide

Learn how to securely receive and verify webhooks in Python using Flask and FastAPI. Covers HMAC signature verification, timestamp validation, and common security pitfalls to avoid.

Send Webhooks with Python: A Complete Tutorial

Learn how to send webhooks with Python using the requests library. This hands-on tutorial covers payload structure, HMAC signatures, error handling, retry logic with exponential backoff, and async options.

Webhook Security Best Practices: The Complete Guide

Learn how to secure your webhook implementations with HMAC signature verification, replay attack prevention, SSRF mitigation, and more. Includes code examples in Node.js and Python.

Webhook Idempotency: Why It Matters and How to Implement It

A comprehensive technical guide to implementing idempotency for webhooks. Learn about idempotency keys, deduplication strategies, and implementation patterns with Node.js and Python code examples.

How to Add Webhooks to Your SaaS Product in 2026

A complete guide for SaaS founders and engineers on implementing webhooks—from event design and payload structure to build vs buy decisions and customer experience best practices.