feat(v0.1.0): project foundation with logging and config
This commit is contained in:
153
src/config/config_loader.py
Normal file
153
src/config/config_loader.py
Normal file
@@ -0,0 +1,153 @@
|
||||
"""Configuration loader with YAML and environment variable support."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import yaml
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from src.core.constants import PATHS
|
||||
from src.core.exceptions import ConfigurationError
|
||||
from src.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# Global config cache
|
||||
_config: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
def load_config(config_path: Optional[Path] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Load configuration from YAML files and environment variables.
|
||||
|
||||
Args:
|
||||
config_path: Path to main config file (defaults to config/config.yaml)
|
||||
|
||||
Returns:
|
||||
Merged configuration dictionary
|
||||
|
||||
Raises:
|
||||
ConfigurationError: If configuration cannot be loaded
|
||||
"""
|
||||
global _config
|
||||
|
||||
if _config is not None:
|
||||
return _config
|
||||
|
||||
# Load environment variables
|
||||
env_path = PATHS["config"].parent / ".env"
|
||||
if env_path.exists():
|
||||
load_dotenv(env_path)
|
||||
logger.debug(f"Loaded environment variables from {env_path}")
|
||||
|
||||
# Load main config
|
||||
if config_path is None:
|
||||
config_path = PATHS["config"] / "config.yaml"
|
||||
|
||||
if not config_path.exists():
|
||||
raise ConfigurationError(
|
||||
f"Configuration file not found: {config_path}",
|
||||
context={"config_path": str(config_path)},
|
||||
)
|
||||
|
||||
try:
|
||||
with open(config_path, "r") as f:
|
||||
config = yaml.safe_load(f) or {}
|
||||
|
||||
# Substitute environment variables
|
||||
config = _substitute_env_vars(config)
|
||||
|
||||
# Load additional config files
|
||||
config_dir = config_path.parent
|
||||
additional_configs = [
|
||||
"logging.yaml",
|
||||
"detectors.yaml",
|
||||
"models.yaml",
|
||||
"trading.yaml",
|
||||
"alerts.yaml",
|
||||
"database.yaml",
|
||||
]
|
||||
|
||||
for config_file in additional_configs:
|
||||
config_file_path = config_dir / config_file
|
||||
if config_file_path.exists():
|
||||
with open(config_file_path, "r") as f:
|
||||
section_config = yaml.safe_load(f) or {}
|
||||
section_config = _substitute_env_vars(section_config)
|
||||
# Merge into main config
|
||||
section_name = config_file.replace(".yaml", "")
|
||||
config[section_name] = section_config
|
||||
|
||||
_config = config
|
||||
logger.info("Configuration loaded successfully")
|
||||
return config
|
||||
|
||||
except Exception as e:
|
||||
raise ConfigurationError(
|
||||
f"Failed to load configuration: {e}",
|
||||
context={"config_path": str(config_path)},
|
||||
) from e
|
||||
|
||||
|
||||
def get_config(key: Optional[str] = None, default: Any = None) -> Any:
|
||||
"""
|
||||
Get configuration value by key (dot-separated path).
|
||||
|
||||
Args:
|
||||
key: Configuration key (e.g., "trading.session.start_time")
|
||||
default: Default value if key not found
|
||||
|
||||
Returns:
|
||||
Configuration value or default
|
||||
"""
|
||||
if _config is None:
|
||||
load_config()
|
||||
|
||||
if key is None:
|
||||
return _config
|
||||
|
||||
keys = key.split(".")
|
||||
value = _config
|
||||
|
||||
for k in keys:
|
||||
if isinstance(value, dict) and k in value:
|
||||
value = value[k]
|
||||
else:
|
||||
return default
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def _substitute_env_vars(config: Any) -> Any:
|
||||
"""
|
||||
Recursively substitute environment variables in config.
|
||||
|
||||
Args:
|
||||
config: Configuration object (dict, list, or primitive)
|
||||
|
||||
Returns:
|
||||
Configuration with environment variables substituted
|
||||
"""
|
||||
if isinstance(config, dict):
|
||||
return {k: _substitute_env_vars(v) for k, v in config.items()}
|
||||
elif isinstance(config, list):
|
||||
return [_substitute_env_vars(item) for item in config]
|
||||
elif isinstance(config, str):
|
||||
# Check for ${VAR} or ${VAR:-default} pattern
|
||||
if config.startswith("${") and config.endswith("}"):
|
||||
var_expr = config[2:-1]
|
||||
if ":-" in var_expr:
|
||||
var_name, default_value = var_expr.split(":-", 1)
|
||||
return os.getenv(var_name.strip(), default_value.strip())
|
||||
else:
|
||||
var_name = var_expr.strip()
|
||||
value = os.getenv(var_name)
|
||||
if value is None:
|
||||
logger.warning(f"Environment variable {var_name} not set")
|
||||
return config # Return original if not found
|
||||
return value
|
||||
return config
|
||||
else:
|
||||
return config
|
||||
|
||||
Reference in New Issue
Block a user