Handle Paystack Webhooks in Django REST Framework
To handle Paystack webhooks in DRF, create an APIView with authentication_classes = [] and permission_classes = [AllowAny]. Read request.body for the raw bytes, compute the HMAC SHA-512 hash, compare it to the x-paystack-signature header, and process the event. Return Response(status=200).
Webhook APIView
# payments/views.py
import json
import hmac
import hashlib
from django.conf import settings
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import AllowAny
class PaystackWebhookView(APIView):
authentication_classes = []
permission_classes = [AllowAny]
def post(self, request):
# 1. Get signature
signature = request.META.get('HTTP_X_PAYSTACK_SIGNATURE', '')
if not signature:
return Response({'error': 'Missing signature'}, status=400)
# 2. Compute hash
secret = settings.PAYSTACK_SECRET_KEY.encode('utf-8')
computed_hash = hmac.new(
secret,
msg=request.body,
digestmod=hashlib.sha512,
).hexdigest()
# 3. Compare
if not hmac.compare_digest(computed_hash, signature):
return Response({'error': 'Invalid signature'}, status=400)
# 4. Process event
event = json.loads(request.body)
self.process_event(event)
return Response(status=200)
def process_event(self, event):
from .services import verify_and_fulfill_order
event_type = event.get('event')
data = event.get('data', {})
if event_type == 'charge.success':
reference = data.get('reference')
if reference:
verify_and_fulfill_order(reference)
elif event_type == 'subscription.create':
self.handle_subscription_create(data)
elif event_type == 'invoice.payment_failed':
self.handle_invoice_failed(data)
def handle_subscription_create(self, data):
pass # Implement based on your subscription model
def handle_invoice_failed(self, data):
pass # Implement based on your subscription modelURL Configuration
# payments/urls.py
from django.urls import path
from .views import InitializePaymentView, VerifyPaymentView, PaystackWebhookView
urlpatterns = [
path('initialize/', InitializePaymentView.as_view()),
path('verify/', VerifyPaymentView.as_view()),
path('webhooks/paystack/', PaystackWebhookView.as_view()),
]Set the webhook URL in your Paystack dashboard: https://yourdomain.com/api/payments/webhooks/paystack/
Log Webhook Events
# payments/models.py
class WebhookEvent(models.Model):
event_type = models.CharField(max_length=100)
reference = models.CharField(max_length=100, null=True, blank=True)
payload = models.JSONField()
processed = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at']Save every event before processing:
def process_event(self, event):
from .models import WebhookEvent
WebhookEvent.objects.create(
event_type=event.get('event', ''),
reference=event.get('data', {}).get('reference', ''),
payload=event,
)
# ... rest of processingThis gives you an audit trail and makes it easy to debug missing payments.
Async Processing with Celery
# payments/tasks.py
from celery import shared_task
from .services import verify_and_fulfill_order
@shared_task
def process_webhook_event(event):
event_type = event.get('event')
data = event.get('data', {})
if event_type == 'charge.success':
reference = data.get('reference')
if reference:
verify_and_fulfill_order(reference)
# In the webhook view:
def process_event(self, event):
from .tasks import process_webhook_event
process_webhook_event.delay(event)This ensures Paystack gets a 200 within milliseconds while your processing happens in a background worker.
Testing
# payments/tests.py
import json
import hmac
import hashlib
from django.conf import settings
from rest_framework.test import APITestCase
class WebhookTest(APITestCase):
def test_valid_webhook(self):
payload = json.dumps({
'event': 'charge.success',
'data': {'reference': 'test_123', 'amount': 500000, 'status': 'success'},
})
signature = hmac.new(
settings.PAYSTACK_SECRET_KEY.encode('utf-8'),
msg=payload.encode('utf-8'),
digestmod=hashlib.sha512,
).hexdigest()
response = self.client.post(
'/api/payments/webhooks/paystack/',
data=payload,
content_type='application/json',
HTTP_X_PAYSTACK_SIGNATURE=signature,
)
self.assertEqual(response.status_code, 200)
def test_invalid_signature(self):
response = self.client.post(
'/api/payments/webhooks/paystack/',
data='{}',
content_type='application/json',
HTTP_X_PAYSTACK_SIGNATURE='invalid',
)
self.assertEqual(response.status_code, 400)Key Takeaways
- ✓Set authentication_classes = [] and permission_classes = [AllowAny] on the webhook APIView. Paystack cannot authenticate with your API.
- ✓DRF does not parse request.body automatically for non-form content types. request.body gives you the raw bytes needed for HMAC.
- ✓Use hmac.compare_digest for constant-time comparison to prevent timing attacks.
- ✓Return Response(status=200) immediately. Process events asynchronously with Celery if needed.
- ✓Make handlers idempotent. Paystack retries failed deliveries and may send the same event multiple times.
- ✓Log every webhook event in a database table for debugging and audit purposes.
Frequently Asked Questions
- Do I need @csrf_exempt with DRF APIView?
- No. DRF APIViews are already CSRF-exempt by default because they use session authentication or token authentication, not Django form-based CSRF. Setting authentication_classes = [] is sufficient.
- Why not use DRF parsers for the webhook body?
- DRF JSONParser would parse the body correctly, but you need the raw bytes for HMAC verification. request.body gives you the raw bytes even when DRF also parses the JSON into request.data.
- Can I use DRF permissions to verify the webhook signature?
- You could create a custom permission class that checks the signature, but it is more common to do it in the view itself since you need access to request.body and the secret key.
Ready to build real-world apps?
Join the McTaba Labs full-stack marathon (4 months full-time · 6 months part-time). Learn M-Pesa, USSD, and WhatsApp engineering while shipping 8 production apps.
Apply to the McTaba Marathon