payment-platform/backend/docs/支付宝SDK封装方案.md

支付宝 SDK 封装方案

一、现状分析

1.1 项目技术栈

组件	技术选型
后端框架	FastAPI 0.115.2
数据库	PostgreSQL 14+ (asyncpg) / MySQL (asyncmy)
ORM	SQLAlchemy 2.0.45
支付宝SDK	alipay-sdk-python >= 3.7.1018
异步框架	uvicorn + asyncio
日志	loguru

1.2 官方 SDK 结构

alipay-sdk-python >= 3.7.1018
├── alipay.aop.api.AlipayClientConfig          # 配置类
├── alipay.aop.api.DefaultAlipayClient         # 客户端类
├── alipay.aop.api.domain.*                     # 业务模型类
├── alipay.aop.api.request.*                    # 请求类
├── alipay.aop.api.response.*                   # 响应类
└── alipay.aop.api.util.SignatureUtils          # 签名验签工具

---

二、目录结构

app/core/alipay/                              # 支付宝 SDK 封装
├── __init__.py                               # 导出接口
├── config.py                                 # SDK 配置类
├── client.py                                 # 统一客户端
└── schema.py                                  # 响应模型定义（基础模型，暂未使用）

---

三、核心组件

3.1 配置类 config.py

class AlipayConfig(BaseSettings):
    """支付宝SDK配置"""

    app_id: str = Field(default="", validation_alias=AliasChoices("ALIPAY_APPID", "ALIPAY_APP_ID"))
    app_private_key: str = Field(default="")
    alipay_public_key: str = Field(default="")
    format: str = Field(default="json")
    charset: str = Field(default="UTF-8")
    sign_type: str = Field(default="RSA2")
    sandbox: bool = Field(default=False)
    notify_url: str = Field(default="")
    return_url: str = Field(default="")
    max_retries: int = Field(default=3)
    request_timeout: int = Field(default=30)
    rate_limit: int = Field(default=100)

    @property
    def is_valid(self) -> bool:
        """检查配置是否有效"""
        return bool(self.app_id and self.app_private_key and self.alipay_public_key)

    def to_alipay_client_config(self) -> "AlipayClientConfig":
        """转换为官方 AlipayClientConfig"""
        config = AlipayClientConfig(sandbox_debug=self.sandbox)
        config.app_id = self.app_id
        config.app_private_key = self.app_private_key
        config.alipay_public_key = self.alipay_public_key
        config.format = self.format
        config.charset = self.charset
        config.sign_type = self.sign_type
        return config


@lru_cache(maxsize=1)
def get_alipay_config() -> AlipayConfig:
    """获取支付宝配置单例"""
    return AlipayConfig()

3.2 客户端类 client.py

class AlipayClient:
    """支付宝客户端工具类"""

    _client: DefaultAlipayClient | None = None

    @classmethod
    def get_client(cls) -> DefaultAlipayClient:
        """获取支付宝客户端实例"""
        if cls._client is None:
            cls._client = DefaultAlipayClient(
                alipay_client_config=get_alipay_config().to_alipay_client_config(),
                logger=logger,
            )
        return cls._client

3.3 响应码枚举 schema.py

AlipayResponseCode = NamedTuple("AlipayResponseCode", [("code", str), ("msg", str)])


class AlipayResponseCodeEnum(Enum):
    """支付宝响应码枚举"""

    SUCCESS = AlipayResponseCode("10000", "接口调用成功")
    SERVICE_UNAVAILABLE = AlipayResponseCode("20000", "服务不可用")
    AUTH_INSUFFICIENT = AlipayResponseCode("20001", "授权权限不足")
    MISSING_REQUIRED_PARAM = AlipayResponseCode("40001", "缺少必选参数")
    INVALID_PARAM = AlipayResponseCode("40002", "非法的参数")
    CONDITION_ABNORMAL = AlipayResponseCode("40003", "条件异常")
    SERVICE_NOT_EXIST = AlipayResponseCode("40004", "服务不存在")
    METHOD_NOT_SUPPORTED = AlipayResponseCode("40005", "不支持请求方式")
    PERMISSION_DENIED = AlipayResponseCode("40006", "权限不足")

    @classmethod
    def from_code(cls, code: str) -> AlipayResponseCode:
        """根据 code 获取对应的 AlipayResponseCode"""
        for item in cls:
            if item.value.code == code:
                return item.value
        return AlipayResponseCode(code, "未知错误")

---

四、接口导出

# app/core/alipay/__init__.py

from app.core.alipay.client import AlipayClient
from app.core.alipay.config import AlipayConfig, get_alipay_config

__all__ = [
    "AlipayConfig",
    "get_alipay_config",
    "AlipayClient",
]

---

五、使用示例

5.1 基础调用

from app.core.alipay import AlipayClient

client = AlipayClient.get_client()

# 构建请求
model = AlipayCommerceEcEnterpriseRegisterinviteCreateModel()
model.out_biz_no = "2024051000000001"
model.identity_type = "ALIPAY_USER_ID"
model.identity = "2088051553855663"
model.register_mode = "NORMAL"
model.sign_fund_way = "BALANCE"

request = AlipayCommerceEcEnterpriseRegisterinviteCreateRequest()
request.biz_model = model

# 执行调用
response = client.execute(request)

# 解析响应
result = AlipayCommerceEcEnterpriseRegisterinviteCreateResponse()
result.parse_response_content(response)

if result.is_success():
    print(result.pc_invite_url)
    print(result.expire_time)

5.2 响应码使用

from app.core.alipay.schema import AlipayResponseCodeEnum

# 获取成功码信息
success_code = AlipayResponseCodeEnum.SUCCESS.value
print(success_code.code)  # "10000"
print(success_code.msg)   # "接口调用成功"

# 根据 code 反查
result = AlipayResponseCodeEnum.from_code("40001")
print(result.code)  # "40001"
print(result.msg)   # "缺少必选参数"

---

六、通知验签

6.1 验签工具

from alipay.aop.api.util.SignatureUtils import verify_with_rsa

# 验签
verify_result = verify_with_rsa(alipay_public_key, biz_content, sign)

6.2 验签流程

支付宝POST请求
    ↓
解析 biz_content
    ↓
使用 verify_with_rsa 验签
    ↓ (失败 → return "fail")
    ↓ (成功 → 继续)
业务处理
    ↓
返回 "success" 或 "fail"

---

七、注意事项

1. 密钥安全：私钥通过环境变量注入，严禁硬编码
2. 沙箱环境：开发时设置 sandbox=True
3. 响应解析：使用 SDK 响应类的 parse_response_content() 方法
4. 验签：使用 verify_with_rsa 方法，自动处理 RSA/RSA2
5. 单例模式：AlipayClient 和 AlipayConfig 都是单例