Verify Paystack Payments in FastAPI

# services.py
import httpx
from decouple import config

PAYSTACK_SECRET = config('PAYSTACK_SECRET_KEY')

async def verify_paystack_transaction(reference: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            'https://api.paystack.co/transaction/verify/' + reference,
            headers={'Authorization': 'Bearer ' + PAYSTACK_SECRET},
        )
    return response.json()

async def verify_and_fulfill(reference: str, db) -> dict:
    order = await db.get_order(reference)
    if not order:
        return {'success': False, 'error': 'Order not found'}

    if order.paid:
        return {'success': True, 'already_fulfilled': True}

    result = await verify_paystack_transaction(reference)

    if not result.get('status') or result['data']['status'] != 'success':
        return {'success': False, 'error': 'Payment not successful'}

    if result['data']['amount'] != order.amount_in_kobo:
        return {'success': False, 'error': 'Amount mismatch'}

    if result['data']['currency'] != order.currency:
        return {'success': False, 'error': 'Currency mismatch'}

    updated = await db.fulfill_order(reference)
    if not updated:
        return {'success': True, 'already_fulfilled': True}

    return {
        'success': True,
        'already_fulfilled': False,
        'amount': order.amount_in_kobo / 100,
        'currency': order.currency,
    }

# main.py
from fastapi import FastAPI, HTTPException, Depends

@app.get('/api/verify')
async def verify_payment(reference: str):
    result = await verify_and_fulfill(reference, db)

    if not result['success']:
        raise HTTPException(status_code=400, detail=result['error'])

    return result

# dependencies.py
from fastapi import Query, HTTPException

async def verified_payment(reference: str = Query(...)):
    result = await verify_paystack_transaction(reference)

    if not result.get('status') or result['data']['status'] != 'success':
        raise HTTPException(status_code=400, detail='Payment not verified')

    return result['data']

# Usage in an endpoint:
@app.get('/api/payment/confirmed')
async def payment_confirmed(payment_data: dict = Depends(verified_payment)):
    return {
        'amount': payment_data['amount'] / 100,
        'currency': payment_data['currency'],
        'email': payment_data['customer']['email'],
    }

import asyncio

async def verify_with_retry(reference: str, max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        try:
            result = await verify_paystack_transaction(reference)
            if result.get('status') is not None:
                return result
        except httpx.RequestError:
            pass

        if attempt < max_retries - 1:
            await asyncio.sleep(2 ** attempt)

    return None

FastAPI async functions work naturally with asyncio.sleep. The event loop is not blocked during the wait.

• Using requests instead of httpx. The requests library blocks the event loop in async endpoints. Use httpx.AsyncClient.
• Not validating the amount. Always compare the Paystack amount against your database, not the client.
• Creating a new AsyncClient per request. For production, use a shared client with connection pooling.
• Mixing test and live keys. Both initialization and verification must use the same mode.

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

Sign up for the McTaba newsletter