如何在Django应用中保存并查看Mpesa till number的交易记录

1. 先搞定Mpesa开发者平台配置 #

先去Safaricom开发者门户注册应用，拿到Consumer Key和Consumer Secret，然后开启C2B交易通知或者交易状态查询的API权限——这是基础，没这个啥都干不了。

2. 两种核心实现方式 #

方式一：实时接收交易推送（推荐） #

Mpesa支持给你的Till Number配置回调URL，用户一完成支付，Mpesa就会把交易详情自动POST到这个地址。你只需要在Django里写个视图接收并存库：

# views.py
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
import json
from .models import MpesaTransaction

@csrf_exempt
def mpesa_callback(request):
    if request.method != 'POST':
        return HttpResponse(status=405)
    
    try:
        data = json.loads(request.body)
        # 按Mpesa返回的字段结构提取信息，字段名可能随环境微调
        transaction, created = MpesaTransaction.objects.get_or_create(
            transaction_id=data['TransID'],
            defaults={
                'till_number': data['BusinessShortCode'],
                'amount': data['TransAmount'],
                'phone_number': data['MSISDN'],
                'transaction_time': data['TransTime'],
                'status': 'completed'
            }
        )
        # 必须返回Mpesa要求的成功响应，不然它会反复推送
        return HttpResponse(json.dumps({"ResultCode": 0, "ResultDesc": "Success"}), content_type='application/json')
    except Exception as e:
        # 可以把错误日志记录下来方便排查
        return HttpResponse(json.dumps({"ResultCode": 1, "ResultDesc": "Failed"}), content_type='application/json')

然后在urls.py加路由：

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('mpesa/callback/', views.mpesa_callback, name='mpesa_callback'),
]

注意：这个回调URL必须是公网能访问的——开发阶段可以用ngrok把本地端口映射成公网HTTPS地址，上线后用正式域名+HTTPS（Mpesa要求回调必须是HTTPS）。最后去Mpesa开发者平台的C2B配置里填这个URL就行。

方式二：主动拉取历史交易 #

如果需要批量查旧交易，或者实时推送没覆盖到的情况，可以用Mpesa的交易状态API主动查询。先写个工具函数拿access token：

# utils.py
import requests
from django.conf import settings

def get_mpesa_token():
    url = "https://sandbox.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials"
    resp = requests.get(url, auth=(settings.MPESA_CONSUMER_KEY, settings.MPESA_CONSUMER_SECRET))
    return resp.json()['access_token']

然后写查询函数：

def fetch_till_transactions(till_num, start_date, end_date):
    token = get_mpesa_token()
    url = "https://sandbox.safaricom.co.ke/mpesa/transactionstatus/v1/query"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    payload = {
        "Initiator": "你的发起者用户名",
        "SecurityCredential": "你的安全凭证（从开发者平台生成）",
        "CommandID": "TransactionStatusQuery",
        "PartyA": till_num,
        "IdentifierType": "4",  # 4代表Till Number
        "ResultURL": "https://你的域名/mpesa/result/",  # 接收查询结果的回调
        "QueueTimeOutURL": "https://你的域名/mpesa/timeout/",
        "Remarks": "拉取Till交易记录"
    }
    resp = requests.post(url, json=payload, headers=headers)
    # 处理返回结果，把交易数据存到数据库
    return resp.json()

可以用django-crontab这类工具做定时任务，每天自动拉取前一天的交易记录。

3. 数据库模型设计 #

建个模型存交易数据，字段按需调整：

# models.py
from django.db import models

class MpesaTransaction(models.Model):
    till_number = models.CharField(max_length=20, verbose_name='Till号码')
    transaction_id = models.CharField(max_length=50, unique=True, verbose_name='交易ID')
    amount = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='交易金额')
    phone_number = models.CharField(max_length=20, verbose_name='付款手机号')
    transaction_time = models.DateTimeField(verbose_name='交易时间')
    status = models.CharField(max_length=20, choices=[('completed', '已完成'), ('failed', '失败')], verbose_name='交易状态')
    created_at = models.DateTimeField(auto_now_add=True, verbose_name='记录创建时间')

    def __str__(self):
        return f"{self.transaction_id} - {self.amount}"

4. 踩坑提醒 #

• 测试用沙箱环境，上线前把API地址换成生产环境的
• 敏感信息（Consumer Key、Secret等）要存在settings.py里，用环境变量或者django-environ管理，别硬编码
• 回调视图要处理重复推送的情况，用get_or_create避免重复存相同交易
• 生产环境必须开HTTPS，不然Mpesa不会给你推送数据

内容的提问来源于stack exchange，提问作者Ahmed Sereta