# 支付宝 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` ```python 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` ```python 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` ```python 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, "未知错误") ``` --- ## 四、接口导出 ```python # 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 基础调用 ```python 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 响应码使用 ```python 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 验签工具 ```python 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` 都是单例