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).
# 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 model
# 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/
Get weekly developer tips
Join 25,000+ developers. Practical guides, job tips, and new content — straight to your inbox.
No spam. Unsubscribe anytime.
# 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 processing
This gives you an audit trail and makes it easy to debug missing payments.
# 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.
# 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)
KES 2,999
The definitive guide to building payment systems with Paystack — from your first transaction to production-grade payment infrastructure across Africa. 9 modules, 47 lessons. Free preview available.
Start Paystack Course