epay-sdk 0.4.0

A production-oriented Python SDK for common EPay-style payment gateways

epay-sdk

一个面向 Python 的易支付 SDK，重点放在“可上线的支付链路”而不只是拼出下单参数。

当前已提供：

• MD5 签名与验签
• 创建订单 URL / 原始下单请求
• 查询订单与结果解析
• 回调验签
• 回调后二次查单确认
• 原子幂等仓储协议
• 默认 SQLAlchemy 仓储（可选安装）
• Decimal 金额标准化
• 作为库时不主动接管业务日志输出

安装

核心功能：

pip install epay-sdk

如需默认 SQLAlchemy 仓储与 ORM 模型：

pip install epay-sdk[sqlalchemy]

如需使用 Django 示例中的集成模板：

pip install epay-sdk[django]

本地开发：

pip install -e .[dev]

快速开始

from epay_sdk import EPayClient, EPayConfig, PayType, PaymentService

client = EPayClient(
    EPayConfig(
        pid="1001",
        key="your_secret_key",
        base_url="https://your-epay.com",
        environment="production",
    )
)
service = PaymentService(client)

pay_url = service.create_order(
    pay_type=PayType.ALIPAY,
    order_no="ORDER_10001",
    amount="9.90",
    subject="会员充值",
    notify_url="https://api.example.com/pay/notify",
    return_url="https://www.example.com/pay/success",
)
print(pay_url)

PaymentService 还保留了便捷方法：

• create_alipay_order(...)
• create_wxpay_order(...)
• create_qqpay_order(...)

create_order、create_order_url、create_order(client) 的区别

• PaymentService.create_order(...)：面向业务层的默认入口。负责金额标准化，并返回可直接跳转/展示给前端的支付 URL。
• EPayClient.create_order_url(CreateOrderRequest)：更底层；当你已经自己构造了 CreateOrderRequest 时使用。
• EPayClient.create_order(CreateOrderRequest)：直接向网关发起请求并返回原始响应文本，只在你确实需要网关原始返回时使用。

如果你的场景只是“生成付款链接”，优先用 PaymentService.create_order(...) 或相应便捷方法。

推荐回调流程

推荐把异步通知处理为如下链路：

1. 收到网关回调表单
2. client.verify_callback() 验签，并校验 pid、sign_type、必要字段、金额格式
3. CallbackProcessor.process() 加载本地订单并校验金额
4. 如启用 verify_with_query=True，自动执行 query_order() + parse_query_result() 做二次确认
5. 通过 mark_paid_if_unpaid() 做原子状态更新
6. 通过 record_callback_attempt(..., stage=...) 记录审计阶段
7. 由调用方决定是否提交/回滚当前事务
8. 首次成功返回 success；失败返回 fail 以便平台重试

回调审计阶段目前包括：

• RECEIVED
• RECONCILED
• APPLIED
• REJECTED

仓储接口约定

你需要自己实现订单仓储，但推荐遵守下面的协议：

class OrderRepository(Protocol):
    def get_order(self, order_no: str) -> Optional[OrderRecord]: ...
    def mark_paid_if_unpaid(self, order_no: str, gateway_trade_no: str, raw_payload: str) -> bool: ...
    def record_callback_attempt(
        self,
        order_no: str,
        accepted: bool,
        reason: Optional[str],
        raw_payload: str,
        *,
        stage: str,
    ) -> None: ...

其中：

• mark_paid_if_unpaid() 必须保证原子性，通常用数据库条件更新实现
• record_callback_attempt() 的 stage 是关键字段，调用方应完整记录
• 如果你的仓储具备事务能力，可额外提供 commit() / rollback()
• CallbackProcessor.process() 默认不提交或回滚外部事务；这样可以避免意外提交同一事务里的其它 ORM 改动
• 如果你明确希望沿用“由回调处理器负责结束事务”的模式，可显式传入 CallbackProcessor(..., manage_transaction=True)，此时它才会在结束时调用仓储的 commit() / rollback()

SQLAlchemy 默认实现

安装 epay-sdk[sqlalchemy] 后，顶层模块会按需暴露：

• Base
• PaymentOrderModel
• CallbackAuditModel
• SQLAlchemyOrderRepository
• create_sqlite_engine

示例：

from sqlalchemy.orm import Session
from epay_sdk import Base, PaymentOrderModel, SQLAlchemyOrderRepository, create_sqlite_engine

engine = create_sqlite_engine("sqlite:///./epay.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(PaymentOrderModel(order_no="A001", amount="9.90", status="UNPAID"))
    session.commit()

    repo = SQLAlchemyOrderRepository(session)
    if repo.mark_paid_if_unpaid("A001", "G100", '{"trade_no":"G100"}'):
        repo.record_callback_attempt("A001", True, None, '{"trade_no":"G100"}', stage="APPLIED")
        repo.commit()

如果未安装 SQLAlchemy extra，import epay_sdk 仍可正常工作；只有在访问上述 SQLAlchemy 符号时才会提示安装可选依赖。

Django 集成模板

Django 集成采取的策略是：

• src/epay_sdk/ 只保留框架无关的核心能力
• Django ORM、Django view、Django URL 与事务边界放在你的应用层实现
• 仓库内提供 examples/django_app/ 作为可直接参考/复制的 Django 模板

也就是说，Django 支持不会耦合进核心 SDK 包本身；SDK 只要求你在 Django 项目中实现符合协议的仓储与回调接入层。

如果你希望直接运行示例，可安装：

pip install epay-sdk[django]

示例入口见：examples/django_app/

Django 中的事务建议

Django 场景下，推荐把回调处理包在 transaction.atomic() 中，并继续使用 CallbackProcessor 的默认行为：

• CallbackProcessor(..., manage_transaction=False)（默认值）
• 由 Django 的 transaction.atomic() 负责提交/回滚
• Django 仓储本身不需要暴露 commit() / rollback()

示意：

from django.db import transaction

with transaction.atomic():
    callback = client.verify_callback(request.POST.dict())
    repo = DjangoOrderRepository()
    processor = CallbackProcessor(client, repo, verify_with_query=True)
    result = processor.process(callback)
return HttpResponse(result.response_text, content_type="text/plain")

这样可以保持和核心 SDK 一致的事务边界：回调处理器负责业务编排，是否提交整个事务由 Django 应用层决定。

Django 仓储如何映射协议

无论使用 Django ORM 还是其他 ORM，仓储仍应满足同一个 OrderRepository 协议。放到 Django 里，通常对应为：

• get_order(order_no)：从 Django model 读取订单，并映射为 OrderRecord
• mark_paid_if_unpaid(order_no, gateway_trade_no, raw_payload)：使用单条条件更新
• record_callback_attempt(...)：写入回调审计表

其中最关键的是 mark_paid_if_unpaid() 的原子性。在 Django 中，推荐明确写成：

updated = PaymentOrder.objects.filter(
    order_no=order_no,
    status="UNPAID",
).update(
    status="PAID",
    gateway_trade_no=gateway_trade_no,
    raw_payload=raw_payload,
    paid_at=timezone.now(),
)
first_success = updated == 1

这对应核心协议里“只允许 UNPAID -> PAID”的幂等要求，避免用“先查再改”的方式破坏并发安全语义。

Django 示例包含什么

examples/django_app/ 当前展示了：

• Django model 设计
• DjangoOrderRepository 的 duck typing 实现
• PaymentService.create_order(...) 的下单 view 用法
• client.verify_callback(...) + CallbackProcessor.process(...) 的回调处理方式
• 使用 transaction.atomic() 控制回调事务
• 示例 URL wiring 与 Django 测试

安全与运行注意事项

• 签名协议说明：SDK 按易支付常见协议使用“排序后的非空参数 + 商户密钥”做 MD5。这里是为了兼容网关协议，不代表 MD5 适合独立承担现代传输安全责任。
• 生产环境请使用 HTTPS：生产配置应使用 https://... 的 base_url，并保持 verify_ssl=True。SDK 已禁止在 environment="production" 时关闭 SSL 校验。
• SQLite 仅适合开发/测试：示例中的 SQLite 适合本地演示；生产支付回调更适合 PostgreSQL/MySQL 等具备更可靠并发/锁语义的数据库。
• 回调必须验签再处理：不要直接信任回调参数；应始终先调用 verify_callback()。
• environment 字段当前是安全/配置语义字段：它目前只影响校验和告警（例如生产环境 TLS 约束），不会自动切换网关地址或协议行为；可视为保留给部署语义和未来扩展的字段。

示例

• examples/flask_app.py：最小 Flask 集成，使用内存仓储演示回调协议
• examples/fastapi_app.py：FastAPI + SQLAlchemy 示例，并显式把同步 SDK/数据库操作放入线程池，避免误导为真正的异步 I/O
• examples/django_app/：Django 集成模板，演示“核心 SDK 保持通用 + Django 仓储/视图/事务放在应用层”