Paystack Mobile Money in Django: Ghana and Kenya (MTN, M-Pesa)

This guide accepts mobile money in Ghana and Kenya through Paystack’s Charge API, in a Django app: MTN, Telecel and AT Money in Ghana, and M-Pesa and Airtel Money in Kenya. You’ll build:

• a paystack app with a payments model that is saved before each charge,
• a small requests client: amounts in subunits, phone formatting, the charge, verification and the webhook signature check,
• a confirm_payment function that settles payments once, only from verified data,
• a webhook view that checks x-paystack-signature, and a reconcile command, which is how failed payments are found,
• tests that run every case without calling Paystack.

Every code block was run in fresh Django 5.2 and 6.1 projects, and the tests pass on both. API facts come from Paystack’s public documentation, checked in November 2026. The Node.js version of this guide explains the flow in more detail.

Before you start

1. In your Paystack dashboard, copy the test secret key from Settings → API Keys & Webhooks, and set the webhook URL to https://your-domain/webhooks/paystack/. Paystack signs webhooks with the same secret key.
2. Install requests and create the app:

pip install "django>=5.2" requests
python manage.py startapp paystack
export PAYSTACK_SECRET_KEY=sk_test_your_test_secret_key

Add "paystack" to INSTALLED_APPS. The provider codes Paystack documents are mtn, vod (Telecel) and atl (AT Money) for GHS, and mpesa and atl (Airtel Money) for KES.

Settings

# settings.py (add to your existing settings)
import os

PAYSTACK = {
    "BASE_URL": "https://api.paystack.co",
    "SECRET_KEY": os.environ.get("PAYSTACK_SECRET_KEY", ""),
}

The model

You choose the reference and send it with the charge, so a lost response never means a lost payment. The amount is stored in the subunit, exactly as Paystack reports it back.

# paystack/models.py
from django.db import models


class PaystackPayment(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending"
        PAID = "paid"
        FAILED = "failed"
        REVIEW = "review"  # paid, but another amount or currency

    reference = models.CharField(max_length=100, unique=True)  # our reference, sent to Paystack
    order_id = models.CharField(max_length=64, db_index=True)
    amount = models.PositiveBigIntegerField()  # in the subunit: pesewas (GHS) or cents (KES)
    currency = models.CharField(max_length=3)  # GHS or KES
    phone = models.CharField(max_length=10)  # local format, e.g. 0551234987
    provider = models.CharField(max_length=10)  # mtn, vod, atl, mpesa
    status = models.CharField(max_length=10, choices=Status.choices, default=Status.PENDING)
    failure_reason = models.CharField(max_length=190, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.order_id} {self.currency} {self.amount / 100:.2f} ({self.status})"

Run python manage.py makemigrations paystack and python manage.py migrate.

The Paystack client

The client turns network problems and 5xx answers into an ambiguous error: the request may have reached Paystack, so the payment must stay pending and be checked. A 4xx is a clear refusal.

# paystack/client.py
import hashlib
import hmac
import re
from urllib.parse import quote

import requests
from django.conf import settings

# Mobile money provider codes Paystack documents, per currency.
PROVIDERS = {"GHS": ("mtn", "vod", "atl"), "KES": ("mpesa", "atl")}
COUNTRY_CODES = {"GHS": "233", "KES": "254"}


class PaystackError(Exception):
    """ambiguous=True: the request may have reached Paystack (timeout, 5xx). Check; don't assume it failed."""

    def __init__(self, message, ambiguous=False):
        super().__init__(message)
        self.ambiguous = ambiguous


def subunits(amount):
    """'25.50' -> 2550. Paystack amounts are in the subunit: pesewas (GHS) or cents (KES)."""
    match = re.fullmatch(r"(\d{1,9})(?:\.(\d{1,2}))?", str(amount).strip())
    if not match:
        raise ValueError("Enter an amount like 25 or 25.50.")
    return int(match.group(1)) * 100 + int((match.group(2) or "").ljust(2, "0"))


def local_phone(raw, currency):
    """'+233 55 123 4987' -> '0551234987': the local format with a leading 0."""
    cc = COUNTRY_CODES.get(currency)
    if cc is None:
        raise ValueError("Currency must be GHS or KES.")
    digits = "".join(ch for ch in str(raw or "") if ch.isdigit())
    if digits.startswith("00"):
        digits = digits[2:]
    if digits.startswith(cc):
        digits = "0" + digits[len(cc):]
    if not digits.startswith("0"):
        digits = "0" + digits
    if not re.fullmatch(r"0\d{9}", digits):
        raise ValueError("Enter a mobile number like 055 123 4987.")
    return digits


def _send(method, path, json=None):
    cfg = settings.PAYSTACK
    try:
        res = requests.request(
            method, cfg["BASE_URL"] + path, json=json, timeout=30,
            headers={"Authorization": f"Bearer {cfg['SECRET_KEY']}", "Accept": "application/json"},
        )
    except requests.RequestException as exc:
        raise PaystackError(f"Paystack unreachable: {exc}", ambiguous=True) from exc
    try:
        body = res.json()
    except ValueError:
        body = None
    if res.status_code >= 500 or not isinstance(body, dict):
        raise PaystackError(f"Paystack HTTP {res.status_code}", ambiguous=True)
    return res, body


def charge(reference, email, amount, currency, phone, provider):
    """Starts a mobile money charge. Returns Paystack's data: status ("pay_offline"), display_text."""
    res, body = _send("POST", "/charge", {
        "reference": reference, "email": email, "amount": amount, "currency": currency,
        "mobile_money": {"phone": phone, "provider": provider},
    })
    if not res.ok or body.get("status") is not True:
        raise PaystackError(f"Paystack refused the charge: {body.get('message') or res.status_code}")
    return body.get("data") or {}


def verify(reference):
    """The source of truth: the transaction for our reference, or None if Paystack has none (yet)."""
    res, body = _send("GET", f"/transaction/verify/{quote(reference)}")
    data = body.get("data")
    return data if res.ok and body.get("status") is True and isinstance(data, dict) else None


def signature_ok(raw_body, signature):
    """x-paystack-signature is an HMAC SHA512 of the raw request body, keyed with your secret key."""
    secret = settings.PAYSTACK.get("SECRET_KEY") or ""
    if not secret or not signature:
        return False
    expected = hmac.new(secret.encode(), raw_body, hashlib.sha512).hexdigest()
    return hmac.compare_digest(expected, signature)

Three details matter:

• Amounts are in the subunit. GHS 25.50 is 2550 pesewas; KES 100 is 10000 cents. subunits() parses the decimal string, so there’s no floating-point rounding.
• Check data.status, not status. The top-level status: true only says the API call worked.
• Phone numbers go in the local format with a leading 0, as in Paystack’s Ghana example (0551234987). Paystack’s docs show the Kenyan test number as +254 710 000 000; this guide sends Kenyan numbers the same local way, so confirm with the test number before going live.

Confirm payments safely

The webhook and the reconcile command both end up here. It asks Paystack first, then locks the row and changes it only if it’s still pending, so a repeated webhook racing the command can’t settle a payment twice.

# paystack/services.py
import logging
from datetime import timedelta

from django.db import transaction
from django.utils import timezone

from . import client
from .models import PaystackPayment

logger = logging.getLogger(__name__)


def confirm_payment(reference):
    """Ask Paystack for the real status and settle the payment once. Safe to call repeatedly.

    Raises client.PaystackError if Paystack can't be reached; the payment stays pending.
    """
    tx = client.verify(reference)  # network call outside the lock

    with transaction.atomic():
        payment = PaystackPayment.objects.select_for_update().filter(reference=reference).first()
        if payment is None or payment.status != PaystackPayment.Status.PENDING:
            return payment  # unknown, or already settled

        if tx is None or tx.get("reference") != payment.reference:
            # No transaction yet. After an hour there never will be: nothing was charged.
            if payment.created_at < timezone.now() - timedelta(hours=1):
                payment.status = PaystackPayment.Status.FAILED
                payment.failure_reason = "not_found"
                payment.save(update_fields=["status", "failure_reason", "updated_at"])
            return payment

        status = str(tx.get("status") or "").lower()
        if status == "success":
            exact = tx.get("currency") == payment.currency and str(tx.get("amount")) == str(payment.amount)
            payment.status = PaystackPayment.Status.PAID if exact else PaystackPayment.Status.REVIEW
            payment.failure_reason = "" if exact else f"paid {tx.get('amount')} {tx.get('currency')}"[:190]
            payment.save(update_fields=["status", "failure_reason", "updated_at"])
            if exact:
                logger.info("Paystack payment for order %s confirmed", payment.order_id)
                # Fulfil the order here, after the commit: transaction.on_commit(lambda: ...)
        elif status in ("failed", "abandoned", "reversed"):
            payment.status = PaystackPayment.Status.FAILED
            payment.failure_reason = str(tx.get("gateway_response") or status)[:190]
            payment.save(update_fields=["status", "failure_reason", "updated_at"])
        return payment  # ongoing, pending and the rest: reconcile_paystack settles it later

The views

Three endpoints: start a payment, let your page poll your own database, and receive Paystack’s webhook. The webhook checks the signature over request.body, the raw bytes, before reading anything from the JSON.

# paystack/views.py
import json
import logging
import uuid

from django.core.exceptions import ValidationError
from django.core.validators import validate_email
from django.http import JsonResponse
from django.shortcuts import get_object_or_404
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from . import client
from .models import PaystackPayment
from .services import confirm_payment

logger = logging.getLogger(__name__)


def _json(raw):
    try:
        data = json.loads(raw or b"{}")
    except ValueError:
        return {}
    return data if isinstance(data, dict) else {}


@require_POST
def start_payment(request):
    """POST {order_id, amount, currency, provider, phone, email} -> 202 with the reference and what to tell the customer."""
    data = _json(request.body)
    order_id = str(data.get("order_id") or "").strip()[:64]
    currency = str(data.get("currency") or "")
    provider = str(data.get("provider") or "")
    try:
        amount = client.subunits(data.get("amount"))  # take it from your order record in a real app
        phone = client.local_phone(data.get("phone"), currency)
        validate_email(str(data.get("email") or ""))
    except (ValueError, ValidationError) as exc:
        return JsonResponse({"error": str(exc.messages[0] if isinstance(exc, ValidationError) else exc)}, status=422)
    if not order_id or provider not in client.PROVIDERS.get(currency, ()):
        return JsonResponse({"error": "Choose an order and a provider available for that currency."}, status=422)

    # Saved BEFORE the call: if the response is lost, reconcile_paystack still finds it by reference.
    payment = PaystackPayment.objects.create(
        reference=f"order-{order_id}-{uuid.uuid4()}", order_id=order_id, amount=amount,
        currency=currency, phone=phone, provider=provider,
    )
    try:
        result = client.charge(payment.reference, data["email"], amount, currency, phone, provider)
    except client.PaystackError as exc:
        logger.warning("Paystack charge for %s: %s", payment.reference, exc)
        if not exc.ambiguous:
            payment.status = PaystackPayment.Status.FAILED
            payment.failure_reason = str(exc)[:190]
            payment.save(update_fields=["status", "failure_reason", "updated_at"])
            return JsonResponse({"error": "The payment could not be started. Please try again."}, status=502)
        result = {}  # it may have gone through: leave it pending, reconcile settles it

    if result.get("status") == "failed":
        payment.status = PaystackPayment.Status.FAILED
        payment.failure_reason = str(result.get("message") or "failed")[:190]
        payment.save(update_fields=["status", "failure_reason", "updated_at"])
        return JsonResponse({"error": "The payment was declined. Please try again."}, status=502)
    return JsonResponse({
        "reference": payment.reference,
        "message": result.get("display_text") or "Approve the payment on your phone.",
    }, status=202)


@require_GET
def payment_status(request, reference):
    """Your order page polls this, not Paystack."""
    payment = get_object_or_404(PaystackPayment, reference=reference)
    return JsonResponse({"status": payment.status})


@csrf_exempt  # Paystack can't send a CSRF token; the signature protects this view instead
@require_POST
def webhook(request):
    # The signature covers the RAW body: check request.body before parsing anything.
    if not client.signature_ok(request.body, request.headers.get("x-paystack-signature", "")):
        return JsonResponse({"error": "Invalid signature"}, status=401)

    data = _json(request.body).get("data")
    reference = data.get("reference") if isinstance(data, dict) else None
    if isinstance(reference, str) and PaystackPayment.objects.filter(reference=reference).exists():
        try:
            confirm_payment(reference)  # never trust the body; this asks Paystack
        except client.PaystackError as exc:
            logger.warning("Paystack verify for %s failed: %s", reference, exc)
            return JsonResponse({"error": "try again"}, status=503)  # Paystack retries
    return JsonResponse({"ok": True})

# paystack/urls.py
from django.urls import path

from . import views

urlpatterns = [
    path("payments/paystack/", views.start_payment),
    path("payments/paystack/<str:reference>/", views.payment_status),
    path("webhooks/paystack/", views.webhook),
]

Include them in your project’s urls.py with path("", include("paystack.urls")). The start view keeps Django’s CSRF check, so call it from your own pages with the X-CSRFToken header. Show the customer the returned message (Paystack’s display_text); Paystack says they have 180 seconds to approve on their phone.

Check pending payments

Paystack’s webhooks only cover successful payments, so this command is how failures and lost webhooks are found:

# paystack/management/commands/reconcile_paystack.py
from datetime import timedelta

from django.core.management.base import BaseCommand
from django.utils import timezone

from paystack.client import PaystackError
from paystack.models import PaystackPayment
from paystack.services import confirm_payment


class Command(BaseCommand):
    help = "Re-check Paystack payments that are still pending. Paystack only sends webhooks for successful ones."

    def handle(self, *args, **options):
        now = timezone.now()
        pending = PaystackPayment.objects.filter(
            status=PaystackPayment.Status.PENDING,
            created_at__lt=now - timedelta(minutes=3),  # the customer has 180 seconds to approve
            created_at__gt=now - timedelta(days=1),
        ).values_list("reference", flat=True)

        settled = 0
        for reference in pending:
            try:
                payment = confirm_payment(reference)
            except PaystackError as exc:
                self.stderr.write(f"{reference}: {exc}")
                continue
            if payment and payment.status != PaystackPayment.Status.PENDING:
                settled += 1
        self.stdout.write(f"Settled {settled} pending payment(s).")

Create empty __init__.py files in paystack/management/ and paystack/management/commands/, then run the command every five minutes from cron:

*/5 * * * * cd /srv/shop && /srv/shop/.venv/bin/python manage.py reconcile_paystack

Test it without calling Paystack

These tests patch requests with unittest.mock, so they run offline and in CI. They check subunits and phone formats, that the webhook body is never trusted, that a bad or unkeyed signature is refused and CSRF doesn’t block real webhooks, that wrong amounts, currencies and references don’t pay, that failed and abandoned payments are final, that a paid payment never changes, and that timeouts and server errors leave payments pending.

# paystack/tests.py
import hashlib
import hmac
import json
from datetime import timedelta
from io import StringIO
from unittest import mock

import requests
from django.core.management import call_command
from django.test import Client, TestCase, override_settings
from django.utils import timezone

from .client import subunits
from .models import PaystackPayment

SECRET = "sk_test_secret"
CHARGED = {"status": True, "message": "Charge attempted",
           "data": {"status": "pay_offline", "display_text": "Please complete authorization process on your mobile phone"}}


def reply(status, body):
    res = mock.Mock(status_code=status, ok=200 <= status < 300)
    res.json.return_value = body
    return res


@override_settings(PAYSTACK={"BASE_URL": "https://api.paystack.co", "SECRET_KEY": SECRET})
class PaystackTests(TestCase):
    def setUp(self):
        self.tx = None  # what verify returns; None = no such transaction
        self.verify_status = 200
        self.charge = (200, CHARGED)
        self.calls = []
        patcher = mock.patch("paystack.client.requests.request", side_effect=self.fake)
        patcher.start()
        self.addCleanup(patcher.stop)

    def fake(self, method, url, json=None, headers=None, timeout=None):
        self.calls.append({"method": method, "url": url, "json": json, "headers": headers})
        if url.endswith("/charge"):
            if isinstance(self.charge, Exception):
                raise self.charge
            return reply(*self.charge)
        if self.verify_status >= 500:
            return reply(self.verify_status, {})
        if self.tx is None:
            return reply(400, {"status": False, "message": "Transaction reference not found"})
        return reply(200, {"status": True, "message": "Verification successful", "data": self.tx})

    def post(self, **overrides):
        body = {"order_id": "1001", "amount": "25.50", "currency": "GHS", "provider": "mtn",
                "phone": "+233 55 123 4987", "email": "buyer@example.com", **overrides}
        return self.client.post("/payments/paystack/", json.dumps(body), content_type="application/json")

    def start(self, **overrides):
        res = self.post(**overrides)
        self.assertEqual(res.status_code, 202, res.content)
        self.assertEqual(res.json()["message"], "Please complete authorization process on your mobile phone")
        return res.json()["reference"]

    def webhook(self, reference, signature=None, client=None):
        body = json.dumps({"event": "charge.success", "data": {"reference": reference, "status": "success"}}).encode()
        if signature is None:
            signature = hmac.new(SECRET.encode(), body, hashlib.sha512).hexdigest()
        return (client or self.client).post("/webhooks/paystack/", body, content_type="application/json",
                                            HTTP_X_PAYSTACK_SIGNATURE=signature)

    def successful(self, reference, **overrides):
        return {"reference": reference, "amount": 2550, "currency": "GHS", "status": "success",
                "gateway_response": "Approved", **overrides}

    def payment(self, reference=None):
        return PaystackPayment.objects.get(reference=reference) if reference else PaystackPayment.objects.get()

    def test_charge_is_sent_in_subunits_and_settled_from_verify_not_the_webhook(self):
        reference = self.start()
        call = self.calls[0]
        self.assertTrue(call["url"].endswith("/charge"))
        self.assertEqual(call["headers"]["Authorization"], "Bearer sk_test_secret")
        self.assertEqual(call["json"], {"reference": reference, "email": "buyer@example.com", "amount": 2550,
                                        "currency": "GHS", "mobile_money": {"phone": "0551234987", "provider": "mtn"}})

        self.tx = self.successful(reference, status="ongoing")  # the body says success; Paystack doesn't
        self.assertEqual(self.webhook(reference).status_code, 200)
        self.assertEqual(self.payment().status, "pending")

        self.tx = self.successful(reference)
        self.webhook(reference)
        self.assertEqual(self.payment().status, "paid")
        self.assertEqual(self.client.get(f"/payments/paystack/{reference}/").json(), {"status": "paid"})

    def test_webhook_with_a_bad_signature_is_refused(self):
        reference = self.start()
        self.tx = self.successful(reference)
        self.assertEqual(self.webhook(reference, "deadbeef").status_code, 401)
        self.assertEqual(self.webhook(reference, "").status_code, 401)
        self.assertFalse([c for c in self.calls if "/transaction/verify/" in c["url"]])
        self.assertEqual(self.payment().status, "pending")

    def test_no_secret_never_accepts_a_signature(self):
        reference = self.start()
        self.tx = self.successful(reference)
        body = json.dumps({"data": {"reference": reference}}).encode()
        unkeyed = hmac.new(b"", body, hashlib.sha512).hexdigest()
        with override_settings(PAYSTACK={"BASE_URL": "https://api.paystack.co", "SECRET_KEY": ""}):
            res = self.client.post("/webhooks/paystack/", body, content_type="application/json", HTTP_X_PAYSTACK_SIGNATURE=unkeyed)
        self.assertEqual(res.status_code, 401)

    def test_webhook_needs_no_csrf_token(self):
        reference = self.start()
        self.tx = self.successful(reference)
        self.assertEqual(self.webhook(reference, client=Client(enforce_csrf_checks=True)).status_code, 200)
        self.assertEqual(self.payment().status, "paid")

    def test_kenya_mpesa_and_bad_input(self):
        self.assertEqual([subunits("7.5"), subunits("25.50"), subunits("100")], [750, 2550, 10000])
        self.start(currency="KES", provider="mpesa", phone="+254 710 000 000", amount=100)
        self.assertEqual(self.calls[0]["json"]["amount"], 10000)
        self.assertEqual(self.calls[0]["json"]["mobile_money"], {"phone": "0710000000", "provider": "mpesa"})
        for bad in ({"provider": "mpesa"}, {"amount": "1.234"}, {"phone": "123"}, {"currency": "NGN"}, {"email": "x"}):
            with self.subTest(bad=bad):
                self.assertEqual(self.post(**bad).status_code, 422)

    def test_another_amount_or_currency_goes_to_review(self):
        for override in ({"amount": 255}, {"currency": "KES"}):
            with self.subTest(override=override):
                reference = self.start()
                self.tx = self.successful(reference, **override)
                self.webhook(reference)
                self.assertEqual(self.payment(reference).status, "review")

    def test_a_transaction_for_another_reference_is_ignored(self):
        reference = self.start()
        self.tx = self.successful("someone-else")
        self.webhook(reference)
        self.assertEqual(self.payment().status, "pending")

    def test_failed_and_abandoned_are_final_and_paid_never_changes(self):
        for status in ("failed", "abandoned"):
            reference = self.start()
            self.tx = self.successful(reference, status=status, gateway_response="Declined")
            self.webhook(reference)
            self.assertEqual([self.payment(reference).status, self.payment(reference).failure_reason], ["failed", "Declined"])
        paid = self.start()
        self.tx = self.successful(paid)
        self.webhook(paid)
        self.tx = self.successful(paid, status="reversed")  # a later answer can't undo it
        self.webhook(paid)
        self.assertEqual(self.payment(paid).status, "paid")

    def test_reconcile_waits_for_a_missing_transaction_then_fails_it_after_an_hour(self):
        self.start()
        PaystackPayment.objects.update(created_at=timezone.now() - timedelta(minutes=5))
        call_command("reconcile_paystack", stdout=StringIO())
        self.assertEqual(self.payment().status, "pending")
        PaystackPayment.objects.update(created_at=timezone.now() - timedelta(minutes=61))
        call_command("reconcile_paystack", stdout=StringIO())
        self.assertEqual([self.payment().status, self.payment().failure_reason], ["failed", "not_found"])

    def test_server_errors_prove_nothing(self):
        reference = self.start()
        PaystackPayment.objects.update(created_at=timezone.now() - timedelta(hours=2))
        self.verify_status = 502
        call_command("reconcile_paystack", stdout=StringIO(), stderr=StringIO())
        self.assertEqual(self.payment().status, "pending")
        self.assertEqual(self.webhook(reference).status_code, 503)  # so Paystack retries the webhook

    def test_charge_timeout_stays_pending_and_reconcile_settles_it(self):
        self.charge = requests.Timeout("read timed out")
        res = self.post()
        self.assertEqual(res.status_code, 202)
        reference = res.json()["reference"]
        self.assertEqual(self.payment().status, "pending")
        self.tx = self.successful(reference)
        PaystackPayment.objects.update(created_at=timezone.now() - timedelta(minutes=5))
        call_command("reconcile_paystack", stdout=StringIO())
        self.assertEqual(self.payment().status, "paid")

    def test_a_clear_refusal_or_a_failed_charge_fails_at_once(self):
        self.charge = (400, {"status": False, "message": "Invalid phone number"})
        self.assertEqual(self.post().status_code, 502)
        self.charge = (200, {"status": True, "data": {"status": "failed", "message": "Declined"}})
        self.assertEqual(self.post().status_code, 502)
        self.assertEqual(list(PaystackPayment.objects.values_list("status", flat=True)), ["failed", "failed"])

Run them with python manage.py test paystack.

Common errors and fixes

Going to production

• Complete Paystack’s business activation for Ghana or Kenya, then switch to your live secret key.
• Load the amount from your order on the server. Never take it from the request body, as the example does for brevity.
• Use Postgres or MySQL in production, so select_for_update() really locks the row across workers.
• Send one small live payment on each provider you support before launch. Paystack’s test numbers are MTN 055 123 4987 and M-Pesa +254 710 000 000.

See also: mobile money in Kenya, mobile money in Ghana, and the M-Pesa STK push guide for Django if you’d rather go direct to Safaricom.

Sending money back out to wallets? See Paystack mobile money payouts.