config.py 3.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. import os
  2. from functools import lru_cache
  3. from typing import TYPE_CHECKING
  4. from pydantic import AliasChoices, Field
  5. from pydantic_settings import BaseSettings, SettingsConfigDict
  6. from app.config.path_conf import ENV_DIR
  7. if TYPE_CHECKING:
  8. from alipay.aop.api.AlipayClientConfig import AlipayClientConfig
  9. class AlipayConfig(BaseSettings):
  10. """支付宝SDK配置"""
  11. model_config = SettingsConfigDict(
  12. env_file=ENV_DIR / f".env.{os.getenv('ENVIRONMENT')}",
  13. env_file_encoding="utf-8",
  14. extra="ignore",
  15. case_sensitive=True,
  16. )
  17. app_id: str = Field(default="", validation_alias=AliasChoices("ALIPAY_APPID", "ALIPAY_APP_ID"), description="支付宝应用ID")
  18. app_private_key: str = Field(default="", validation_alias=AliasChoices("ALIPAY_PRIVATE_KEY"), description="应用私钥(RSA2)")
  19. alipay_public_key: str = Field(default="", validation_alias=AliasChoices("ALIPAY_PUBLIC_KEY"), description="支付宝公钥(RSA2)")
  20. format: str = Field(default="json", validation_alias=AliasChoices("ALIPAY_FORMAT"), description="响应格式")
  21. charset: str = Field(default="UTF-8", validation_alias=AliasChoices("ALIPAY_CHARSET"), description="字符集")
  22. sign_type: str = Field(default="RSA2", validation_alias=AliasChoices("ALIPAY_SIGN_TYPE"), description="签名类型(RSA/RSA2)")
  23. sandbox: bool = Field(default=False, validation_alias=AliasChoices("ALIPAY_SANDBOX"), description="是否沙箱环境")
  24. notify_url: str = Field(default="", validation_alias=AliasChoices("ALIPAY_NOTIFY_URL"), description="异步通知地址")
  25. return_url: str = Field(default="", validation_alias=AliasChoices("ALIPAY_RETURN_URL"), description="同步跳转地址")
  26. max_retries: int = Field(default=3, description="最大重试次数")
  27. request_timeout: int = Field(default=30, description="请求超时时间(秒)")
  28. rate_limit: int = Field(default=100, description="限流阈值(每分钟)")
  29. # production_server_url: str = Field(default="https://openapi.alipay.com/gateway.do", description="生产环境网关地址")
  30. # sandbox_server_url: str = Field(default="https://openapi-sandbox.dl.alipaydev.com/gateway.do", description="沙箱环境网关地址")
  31. #
  32. # @property
  33. # def server_url(self) -> str:
  34. # """获取支付宝网关地址"""
  35. # if self.sandbox:
  36. # return self.sandbox_server_url
  37. # return self.production_server_url
  38. @property
  39. def is_valid(self) -> bool:
  40. """检查配置是否有效"""
  41. return bool(self.app_id and self.app_private_key and self.alipay_public_key)
  42. def to_alipay_client_config(self) -> "AlipayClientConfig":
  43. """转换为官方AlipayClientConfig"""
  44. from alipay.aop.api.AlipayClientConfig import AlipayClientConfig
  45. config = AlipayClientConfig(sandbox_debug=self.sandbox)
  46. config.app_id = self.app_id
  47. config.app_private_key = self.app_private_key
  48. config.alipay_public_key = self.alipay_public_key
  49. config.format = self.format
  50. config.charset = self.charset
  51. config.sign_type = self.sign_type
  52. return config
  53. @lru_cache(maxsize=1)
  54. def get_alipay_config() -> AlipayConfig:
  55. """获取支付宝配置单例"""
  56. return AlipayConfig()