payments/支付宝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

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.*                   # 响应类

---

二、目录结构

app/core/alipay/                              # 支付宝 SDK 封装
├── __init__.py                               # 导出接口
├── config.py                                 # SDK 配置类
├── client.py                                 # 统一客户端
├── schema.py                                  # 响应模型定义
└── services/                                 # 服务层（规划中）

---

三、核心组件

3.1 配置类 config.py

from pydantic_settings import BaseSettings, SettingsConfigDict

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="")

    @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

3.2 客户端类 client.py

from alipay.aop.api.DefaultAlipayClient import DefaultAlipayClient

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

基础响应模型

class AlipayResponse(BaseModel):
    """支付宝响应基础模型"""

    code: str = Field(..., description="网关返回码，10000表示成功")
    msg: str = Field(..., description="网关返回码描述")
    sub_code: Optional[str] = Field(None, description="业务错误码")
    sub_msg: Optional[str] = Field(None, description="业务错误描述")

    @property
    def success(self) -> bool:
        return self.code == AlipayResponseCodeEnum.SUCCESS.value.code

响应码枚举

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, "未知错误")

业务响应模型示例

class AlipayCommerceEcEnterpriseRegisterInviteCreateResponse(AlipayResponse):
    """邀请企业注册响应"""

    pc_invite_url: Optional[str] = Field(None, description="企业注册/认证/签约三合一页面的链接地址")
    expire_time: Optional[datetime] = Field(None, description="链接过期时间")

    @field_validator("expire_time", mode="before")
    @classmethod
    def parse_expire_time(cls, v: Optional[str]) -> Optional[datetime]:
        if v is None:
            return None
        return datetime.strptime(v, "%Y-%m-%d %H:%M:%S")

---

四、使用示例

4.1 基础调用

from app.core.alipay import AlipayClient, get_alipay_config

# 获取客户端
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)

# 使用 Pydantic 模型验证响应
result = AlipayCommerceEcEnterpriseRegisterInviteCreateResponse.model_validate_json(response)
if result.success:
    print(result.pc_invite_url)
    print(result.expire_time)

4.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)   # "缺少必选参数"

---

五、接口导出

# 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",
]

---

六、规划中功能

6.1 服务层封装

services/
├── base.py                             # 基础服务类
├── enterprise_service.py               # 企业服务
├── employee_service.py                  # 员工服务
├── expense_service.py                  # 费控服务
├── quota_service.py                    # 额度服务
└── bill_service.py                     # 账单服务

6.2 异常体系

class AlipayError(Exception):
    """支付宝基础异常"""
    code: str = "500"
    msg: str = "Unknown error"


class AlipayApiError(AlipayError):
    """API 调用错误"""
    code = "40000"


class AlipayNetworkError(AlipayError):
    """网络错误（可重试）"""
    code = "20000"

6.3 重试机制

基于指数退避的重试策略，支持的错误码：

• 20000 - 服务不可用
• 20001 - 授权权限不足
• ACQ.SystemError - 银行系统异常

---

七、注意事项

1. 密钥安全：私钥通过环境变量注入，严禁硬编码
2. 沙箱环境：开发时设置 sandbox=True
3. 响应验证：使用 Pydantic 模型验证响应数据
4. 错误处理：根据 code 和 sub_code 进行分类处理
5. 超时控制：默认超时 30 秒