config.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172
  1. from pydantic_settings import BaseSettings
  2. from .config_app import AppConfig
  3. from .config_database import DatabaseConfig
  4. from .config_logging import LoggingConfig
  5. class Config(BaseSettings):
  6. app: AppConfig = AppConfig()
  7. database: DatabaseConfig = DatabaseConfig()
  8. logging: LoggingConfig = LoggingConfig()
  9. class Config:
  10. env_prefix = ''
  11. env_nested_delimiter = '__'
  12. case_sensitive = False
  13. def __init__(self, **data):
  14. super().__init__(**data)
  15. # 确保logging配置被正确初始化
  16. if not hasattr(self, 'logging'):
  17. self.logging = LoggingConfig()
  18. def to_string(self, indent: int = 0) -> str:
  19. """将配置转换为字符串"""
  20. indent_str = ' ' * indent
  21. result = []
  22. for field_name, field_value in self.__dict__.items():
  23. if hasattr(field_value, 'to_string'):
  24. result.append(
  25. f"{indent_str}{field_name}:\n{field_value.to_string(indent + 4)}"
  26. )
  27. else:
  28. result.append(f"{indent_str}{field_name}: {field_value}")
  29. return '\n'.join(result)
  30. def _update_config_section(self, config_dict: dict, section: str,
  31. config_class: type) -> None:
  32. """
  33. 更新配置的某个部分
  34. Args:
  35. config_dict: 包含配置项的字典
  36. section: 配置部分名称
  37. config_class: 配置类
  38. """
  39. if section in config_dict:
  40. setattr(
  41. self, section,
  42. config_class(
  43. **{
  44. k: v
  45. for k, v in config_dict[section].items()
  46. if k in config_class.model_fields
  47. }))
  48. def update_from_dict(self,
  49. config_dict: dict,
  50. partial: bool = False) -> None:
  51. """
  52. 从字典更新配置
  53. Args:
  54. config_dict: 包含配置项的字典
  55. partial: 是否部分更新,默认为False
  56. """
  57. try:
  58. self._update_config_section(config_dict, 'app', AppConfig)
  59. self._update_config_section(config_dict, 'database',
  60. DatabaseConfig)
  61. self._update_config_section(config_dict, 'logging', LoggingConfig)
  62. except Exception as e:
  63. raise ValueError(f"配置更新失败: {e}")