McTaba Labs logo
Bonaventure OgetoBy Bonaventure Ogeto|

Handle Paystack Webhooks in FastAPI

To handle Paystack webhooks in FastAPI, create a POST endpoint that reads the raw request body with await request.body(), computes an HMAC SHA-512 hash using your secret key, and compares it to the x-paystack-signature header. Process the event in a BackgroundTask and return 200.

Webhook Endpoint

# main.py
import hmac
import hashlib
import json
from fastapi import FastAPI, Request, BackgroundTasks, HTTPException

app = FastAPI()

PAYSTACK_SECRET = config('PAYSTACK_SECRET_KEY')

@app.post('/api/webhooks/paystack')
async def paystack_webhook(request: Request, background_tasks: BackgroundTasks):
    # 1. Get signature
    signature = request.headers.get('x-paystack-signature', '')

    if not signature:
        raise HTTPException(status_code=400, detail='Missing signature')

    # 2. Get raw body
    body = await request.body()

    # 3. Compute hash
    computed_hash = hmac.new(
        PAYSTACK_SECRET.encode('utf-8'),
        msg=body,
        digestmod=hashlib.sha512,
    ).hexdigest()

    # 4. Compare
    if not hmac.compare_digest(computed_hash, signature):
        raise HTTPException(status_code=400, detail='Invalid signature')

    # 5. Parse and process in background
    event = json.loads(body)
    background_tasks.add_task(process_event, event)

    return {'status': 'ok'}

Process Events

# webhooks.py
async def process_event(event: dict):
    event_type = event.get('event')
    data = event.get('data', {})

    if event_type == 'charge.success':
        reference = data.get('reference')
        if reference:
            await verify_and_fulfill(reference, db)

    elif event_type == 'subscription.create':
        await handle_subscription_create(data)

    elif event_type == 'invoice.payment_failed':
        await handle_invoice_failed(data)

    elif event_type == 'transfer.success':
        await handle_transfer_success(data)

Signature Verification as a Dependency

# dependencies.py
async def verify_paystack_signature(request: Request) -> dict:
    signature = request.headers.get('x-paystack-signature', '')
    if not signature:
        raise HTTPException(status_code=400, detail='Missing signature')

    body = await request.body()

    computed_hash = hmac.new(
        PAYSTACK_SECRET.encode('utf-8'),
        msg=body,
        digestmod=hashlib.sha512,
    ).hexdigest()

    if not hmac.compare_digest(computed_hash, signature):
        raise HTTPException(status_code=400, detail='Invalid signature')

    return json.loads(body)

# Usage:
@app.post('/api/webhooks/paystack')
async def paystack_webhook(
    event: dict = Depends(verify_paystack_signature),
    background_tasks: BackgroundTasks = BackgroundTasks(),
):
    background_tasks.add_task(process_event, event)
    return {'status': 'ok'}

The dependency pattern keeps your endpoint clean and makes the signature verification reusable and testable.

Testing Webhooks

# tests/test_webhook.py
import hmac
import hashlib
import json
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_valid_webhook():
    payload = json.dumps({
        'event': 'charge.success',
        'data': {'reference': 'test_123', 'amount': 500000},
    })

    signature = hmac.new(
        b'sk_test_xxx',
        msg=payload.encode(),
        digestmod=hashlib.sha512,
    ).hexdigest()

    response = client.post(
        '/api/webhooks/paystack',
        content=payload,
        headers={
            'Content-Type': 'application/json',
            'x-paystack-signature': signature,
        },
    )

    assert response.status_code == 200

Production Notes

  1. HTTPS required. Your webhook URL must use HTTPS in production.
  2. Respond quickly. BackgroundTasks run after the response is sent. Paystack gets a fast 200.
  3. For heavy processing. Use Celery or an async task queue instead of BackgroundTasks if processing takes more than a few seconds.
  4. Log events. Store every webhook event in a database for debugging.
  5. Monitor. Check the Paystack dashboard for failed webhook deliveries.

Ship Payments Faster

Key Takeaways

  • ✓Use await request.body() to get the raw bytes for HMAC verification.
  • ✓FastAPI BackgroundTasks let you return 200 immediately and process the event asynchronously.
  • ✓Use hmac.compare_digest for constant-time comparison to prevent timing attacks.
  • ✓Make handlers idempotent. Paystack retries failed webhooks.
  • ✓The most important event is charge.success. Handle subscription events if you use recurring billing.
  • ✓FastAPI does not need CSRF protection since it is an API framework, so no special exemptions are needed.

Frequently Asked Questions

Does FastAPI need CSRF exemption for webhooks?
No. FastAPI is an API framework and does not have CSRF protection by default. Unlike Django, no special decorator is needed.
Should I use BackgroundTasks or Celery for webhook processing?
BackgroundTasks are simpler and work well for lightweight processing. For heavy operations (sending emails, calling external APIs, complex database updates), use Celery with a message broker like Redis.
Can I use the same endpoint for test and live webhooks?
Yes. Use the correct secret key for each mode. Test webhooks are signed with your test key, live webhooks with your live key. If you serve both, determine the mode from the webhook data or use separate keys.

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