clear config

This commit is contained in:
gongwenxin 2025-05-29 16:42:14 +08:00
parent 800efa78ef
commit c3bd515cee
2 changed files with 0 additions and 114 deletions

View File

@ -1 +0,0 @@

View File

@ -1,113 +0,0 @@
"""Configuration Management Module"""
import yaml
from pydantic import ValidationError # BaseModel is not directly used here anymore for defining AppConfig
from typing import Optional, Any
from ..models.config_models import AppConfig # Import AppConfig from the new location
class ConfigurationManager:
"""
Loads and manages application configuration from a YAML file.
Uses Pydantic for validation.
"""
def __init__(self, config_path: str = "configs/config.yaml"):
self.config_path = config_path
self.config: Optional[AppConfig] = None
self._load_config()
def _load_config(self):
try:
with open(self.config_path, 'r') as f:
raw_config = yaml.safe_load(f)
if raw_config is None:
raw_config = {} # Allow empty config file, defaults from Pydantic will be used
self.config = AppConfig(**raw_config)
print(f"Configuration loaded successfully from {self.config_path}")
except FileNotFoundError:
print(f"Warning: Configuration file {self.config_path} not found. Using default settings.")
self.config = AppConfig() # Load with default values
except yaml.YAMLError as e:
print(f"Error parsing YAML configuration file {self.config_path}: {e}")
print("Falling back to default configuration.")
self.config = AppConfig()
except ValidationError as e:
print(f"Configuration validation error from {self.config_path}:\n{e}")
print("Falling back to default configuration due to validation errors.")
self.config = AppConfig()
except Exception as e:
print(f"An unexpected error occurred while loading configuration from {self.config_path}: {e}")
print("Falling back to default configuration.")
self.config = AppConfig()
def get_config(self) -> AppConfig:
"""Returns the loaded (and validated) application configuration."""
if self.config is None:
print("Error: Config not loaded. Attempting to load defaults.")
self.config = AppConfig()
return self.config
def get_module_config(self, module_name: str) -> Optional[Any]:
"""Returns configuration for a specific module."""
if self.config:
# Ensure the module_name is a valid attribute of AppConfig
if hasattr(self.config, module_name):
return getattr(self.config, module_name)
else:
print(f"Warning: Configuration for module '{module_name}' not found in AppConfig.")
return None
return None
# Example Usage (can be moved to tests or main application logic)
if __name__ == '__main__':
# Create a dummy configs directory if it doesn't exist for the example
import os
if not os.path.exists("configs"):
os.makedirs("configs")
dummy_config_file = "configs/dummy_config.yaml"
dummy_config_content = {
'app_name': 'My DDMS Checker',
'logging': {'level': 'DEBUG'},
'api_caller': {
'default_timeout': 60,
'default_headers': {'X-Custom-Header': 'TestValue'}
},
'rule_repository': {
'storage': {
'type': 'filesystem',
'path': './custom_rules'
},
'preload_rules': True
},
'json_schema_validator': { # Added for completeness of the example
'default_draft_version': 'draft7'
}
}
with open(dummy_config_file, 'w') as f_yaml:
yaml.dump(dummy_config_content, f_yaml)
manager = ConfigurationManager(config_path=dummy_config_file)
app_cfg = manager.get_config()
print(f"App Name: {app_cfg.app_name}")
print(f"Log Level: {app_cfg.logging.level}")
if app_cfg.api_caller:
print(f"API Caller Timeout: {app_cfg.api_caller.default_timeout}")
if app_cfg.rule_repository and app_cfg.rule_repository.storage:
print(f"Rule Storage Type: {app_cfg.rule_repository.storage.type}")
print(f"Rule Storage Path: {app_cfg.rule_repository.storage.path}")
if app_cfg.json_schema_validator:
print(f"JSON Schema Validator Draft: {app_cfg.json_schema_validator.default_draft_version}")
print("\nLoading with non-existent file (expect defaults):")
manager_default = ConfigurationManager(config_path="configs/non_existent_config.yaml")
app_cfg_default = manager_default.get_config()
print(f"App Name (Default): {app_cfg_default.app_name}")
if app_cfg_default.api_caller:
print(f"API Caller Timeout (Default): {app_cfg_default.api_caller.default_timeout}")
# Clean up the dummy file and directory if it was created
if os.path.exists(dummy_config_file):
os.remove(dummy_config_file)
if os.path.exists("configs") and not os.listdir("configs"):
os.rmdir("configs")