wechat-xpay 0.2.0

Python SDK for WeChat XPay (Virtual Payment) server-side APIs

微信支付虚拟支付 Python SDK

微信支付虚拟支付（XPay）服务端 API 的 Python SDK。

特性

• ✅ 同步与异步 - 同时支持同步和异步操作
• ✅ 类型提示 - 完整的类型注解支持
• ✅ HTTP/2 - 基于 httpx，支持 HTTP/2
• ✅ Webhook 解析 - 处理微信推送通知
• ✅ 错误处理 - 完善的异常层次结构和错误码

安装

pip install wechat-xpay

快速开始

同步客户端

from wechat_xpay import XPayClient

# 使用上下文管理器（推荐）
with XPayClient(
    app_id="你的_app_id",
    app_key="你的_app_key",
    env=0,  # 0=沙箱环境，1=生产环境
) as client:
    # session_key 在每次调用 API 时传入，因为它会定期过期
    balance = client.query_user_balance(
        openid="用户_openid",
        session_key="用户_session_key",
    )
    print(f"余额: {balance.balance}")
    print(f"赠送余额: {balance.present_balance}")

异步客户端

import asyncio
from wechat_xpay import XPayAsyncClient

async def main():
    # 使用异步上下文管理器（推荐）
    async with XPayAsyncClient(
        app_id="你的_app_id",
        app_key="你的_app_key",
        env=0,
    ) as client:
        balance = await client.query_user_balance(
            openid="用户_openid",
            session_key="用户_session_key",
        )
        print(f"余额: {balance.balance}")
        print(f"赠送余额: {balance.present_balance}")

asyncio.run(main())

为什么 session_key 要在每次调用时传入？

微信的 session_key 有生命周期，会定期过期（通常 30 天）。当它过期时，你需要重新授权用户获取新的 session_key。

通过在每次 API 调用时传入 session_key 而不是在初始化时传入：

• 你可以优雅地处理 session_key 过期
• 不同用户可以使用同一个客户端实例，各自使用自己的 session_key
• 你可以轮换 session_key 而无需重新创建客户端

API 覆盖

• 用户代币管理 (query_user_balance, currency_pay, cancel_currency_pay, present_currency)
• 订单管理 (query_order)
• 退款 (refund_order)
• 提现 (create_withdraw_order, query_withdraw_order)
• 发货通知 (notify_provide_goods)
• 商家余额 (query_biz_balance)
• 广告金 (query_transfer_account, query_adver_funds, create_funds_bill)
• 广告金账单 (query_funds_bill, query_recover_bill, download_adverfunds_order)
• 转账账户绑定 (bind_transfer_account)
• 道具管理 (start_upload_goods, query_upload_goods, start_publish_goods, query_publish_goods)
• 投诉管理 (get_complaint_list, get_complaint_detail, response_complaint, complete_complaint, get_negotiation_history)
• 文件上传 (upload_vp_file, get_upload_file_sign)
• 账单下载 (download_bill, download_adverfunds_order)
• Webhook 解析 (发货通知、代币支付通知、退款通知、投诉通知)

共计 29 个 API 端点，已全部实现。

使用示例

同步使用

from wechat_xpay import XPayClient

with XPayClient(
    app_id="wx1234567890",
    app_key="你的_app_key",
    env=0,
) as client:
    # 查询用户余额
    balance = client.query_user_balance(
        openid="用户_openid",
        session_key="用户_session_key",
    )
    print(f"余额: {balance.balance}")
    print(f"赠送余额: {balance.present_balance}")

    # 处理支付
    result = client.currency_pay(
        openid="用户_openid",
        session_key="用户_session_key",
        order_id="订单_123",
        amount=100,  # 代币数量
        payitem="商品描述",
    )
    print(f"订单 ID: {result.order_id}")

    # 退款
    result = client.refund_order(
        openid="用户_openid",
        session_key="用户_session_key",
        refund_order_id="退款_123",
        left_fee=1000,
        refund_fee=500,
        refund_reason="1",  # 商品问题
        req_from="1",  # 人工
        order_id="原订单_id",
    )
    print(f"退款订单 ID: {result.refund_order_id}")

异步使用

import asyncio
from wechat_xpay import XPayAsyncClient

async def main():
    async with XPayAsyncClient(
        app_id="wx1234567890",
        app_key="你的_app_key",
        env=0,
    ) as client:
        # 查询用户余额
        balance = await client.query_user_balance(
            openid="用户_openid",
            session_key="用户_session_key",
        )
        print(f"余额: {balance.balance}")

        # 并发处理多个支付（使用不同的 session_key）
        tasks = [
            client.currency_pay(
                openid=f"用户_{i}",
                session_key=f"session_key_{i}",  # 每个用户有自己的 session_key
                order_id=f"订单_{i}",
                amount=100,
                payitem=f"商品_{i}",
            )
            for i in range(3)
        ]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(f"订单 ID: {result.order_id}")

asyncio.run(main())

处理 Webhook

from wechat_xpay.webhook import WebhookParser

parser = WebhookParser()

# 解析 JSON 负载
notification = parser.parse({
    "Event": "xpay_goods_deliver_notify",
    "OpenId": "用户_openid",
    "OutTradeNo": "订单_123",
    # ... 其他字段
})

print(f"事件类型: {notification.event}")
print(f"用户 OpenID: {notification.open_id}")

# 返回成功响应给微信
response = parser.success_response()

错误处理

from wechat_xpay import XPayClient
from wechat_xpay.exceptions import XPayAPIError, ERR_SESSION_KEY_EXPIRED

client = XPayClient(...)

try:
    balance = client.query_user_balance(
        openid="用户_openid",
        session_key="用户_session_key",
    )
except XPayAPIError as e:
    if e.errcode == ERR_SESSION_KEY_EXPIRED:
        print("会话已过期，请重新登录")
        # 重新授权用户获取新的 session_key
    else:
        print(f"API 错误: {e.errcode} - {e.errmsg}")

客户端生命周期管理

同步客户端

# 方式 1：上下文管理器（自动关闭）
with XPayClient(...) as client:
    result = client.query_user_balance(...)

# 方式 2：手动关闭
client = XPayClient(...)
try:
    result = client.query_user_balance(...)
finally:
    client.close()

异步客户端

# 方式 1：异步上下文管理器（自动关闭）
async with XPayAsyncClient(...) as client:
    result = await client.query_user_balance(...)

# 方式 2：手动关闭
client = XPayAsyncClient(...)
try:
    result = await client.query_user_balance(...)
finally:
    await client.close()

认证

SDK 自动处理两种签名：

• pay_sig: 使用 AppKey 签名，用于 API 认证（在客户端初始化时配置）
• signature: 使用用户的 session_key 签名，用于用户态验证（每次 API 调用时传入）

环境

• env=0: 沙箱环境（用于测试）
• env=1: 生产环境

文档

查看 docs/plans/ 目录了解实现细节和 API 规范。

许可证

MIT