base.py 2.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. import os
  2. from dotenv import load_dotenv
  3. from core.utils.path_utils import project_root,log_dir
  4. from pydantic import BaseSettings, Field, AnyUrl
  5. # 在 Settings 类之前加载 .env 文件
  6. load_dotenv(os.path.join(project_root,".env"))
  7. class Settings(BaseSettings):
  8. # 环境标识
  9. ENV: str = "prod" # prod/dev
  10. ENABLE_ALIYUN_LOG: bool = True
  11. # 日志配置
  12. LOG_DIR: str = log_dir
  13. LOG_LEVEL: str = "INFO"
  14. # 阿里云数据库配置 (RDS)
  15. DB_HOST: str = Field(..., env="DB_HOST")
  16. DB_PORT: int = Field(3306, env="DB_PORT")
  17. DB_USER: str = Field(..., env="DB_USER")
  18. DB_PASSWORD: str = Field(..., env="DB_PASSWORD")
  19. DB_NAME: str = Field(..., env="DB_NAME")
  20. DB_CHARSET: str = Field(..., env="DB_CHARSET")
  21. DB_POOL_SIZE: int = 20
  22. DB_POOL_RECYCLE: int = 3600
  23. # 阿里云RocketMQ配置
  24. ROCKETMQ_ENDPOINT: AnyUrl = Field(..., env="ROCKETMQ_ENDPOINT")
  25. ROCKETMQ_ACCESS_KEY_ID: str = Field(..., env="ROCKETMQ_ACCESS_KEY_ID")
  26. ROCKETMQ_ACCESS_KEY_SECRET: str = Field(..., env="ROCKETMQ_ACCESS_KEY_SECRET")
  27. ROCKETMQ_INSTANCE_ID: str = Field(..., env="ROCKETMQ_INSTANCE_ID")
  28. ROCKETMQ_WAIT_SECONDS: int = 10
  29. ROCKETMQ_BATCH: int = 1
  30. # 飞书配置
  31. FEISHU_APPID: str = Field(..., env="FEISHU_APPID")
  32. FEISHU_APPSECRET: str = Field(..., env="FEISHU_APPSECRET")
  33. # 连接池配置
  34. CONNECTION_TIMEOUT: int = 10
  35. REQUEST_TIMEOUT: int = 30
  36. # 阿里云日志
  37. ALIYUN_ACCESS_KEY_ID: str = Field(..., env="ALIYUN_ACCESS_KEY_ID")
  38. ALIYUN_ACCESS_KEY_SECRET: str = Field(..., env="ALIYUN_ACCESS_KEY_SECRET")
  39. # redis
  40. REDIS_HOST: str = Field(..., env="REDIS_HOST")
  41. REDIS_PORT: int = Field(..., env="REDIS_PORT")
  42. REDIS_PASSWORD: str = Field(..., env="REDIS_PASSWORD")
  43. REDIS_DB: int = Field(0, env="REDIS_DB")
  44. REDIS_MAX_CONNECTIONS: int = Field(20, env="REDIS_MAX_CONNECTIONS")
  45. @property
  46. def redis_url(self) -> str:
  47. """生成"""
  48. return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
  49. class Config:
  50. env_file = ".env"
  51. env_file_encoding = 'utf-8'
  52. case_sensitive = False
  53. # @property
  54. # def database_url(self) -> str:
  55. # """生成安全的数据库连接字符串"""
  56. # return f"mysql+asyncmy://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}?charset={self.DB_CHARSET}"
  57. settings = Settings()