Handle Paystack Webhooks in FastAPI

# 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'}

# 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)

# 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.

# 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

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.

This guide is part of the Paystack integration guides by language and framework series.

Sign up for the McTaba newsletter