Merge PR #743: config_overlay.py (added)
This commit is contained in:
130
config_overlay.py
Normal file
130
config_overlay.py
Normal file
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
config_overlay.py — Environment-specific config overlays.
|
||||
|
||||
Merges a base config YAML with an environment-specific overlay.
|
||||
Overlay keys override base keys (deep merge for dicts, replace for lists).
|
||||
|
||||
Usage:
|
||||
from config_overlay import load_config
|
||||
config = load_config('config.yaml', env='prod')
|
||||
# Loads config.yaml + config.prod.yaml, merges, returns dict
|
||||
|
||||
Environments: dev, prod, cron, gateway (or custom)
|
||||
"""
|
||||
import os
|
||||
import copy
|
||||
import yaml
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def deep_merge(base: dict, overlay: dict) -> dict:
|
||||
"""Deep merge overlay into base. Overlay wins on conflicts.
|
||||
|
||||
- Dicts are merged recursively
|
||||
- Lists are replaced (not extended)
|
||||
- Scalars are replaced
|
||||
- None in overlay removes the key from base
|
||||
"""
|
||||
result = copy.deepcopy(base)
|
||||
for key, value in overlay.items():
|
||||
if value is None:
|
||||
result.pop(key, None)
|
||||
elif key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
||||
result[key] = deep_merge(result[key], value)
|
||||
else:
|
||||
result[key] = copy.deepcopy(value)
|
||||
return result
|
||||
|
||||
|
||||
def find_config(base_path: str, env: str = None) -> tuple:
|
||||
"""Find base and overlay config paths.
|
||||
|
||||
Returns: (base_path, overlay_path_or_None)
|
||||
"""
|
||||
base = Path(base_path)
|
||||
if not base.exists():
|
||||
raise FileNotFoundError(f"Base config not found: {base_path}")
|
||||
|
||||
overlay = None
|
||||
if env:
|
||||
# Try: config.{env}.yaml, config/{env}.yaml, {env}.overlay.yaml
|
||||
candidates = [
|
||||
base.parent / f"{base.stem}.{env}{base.suffix}",
|
||||
base.parent / env / f"{base.name}",
|
||||
base.parent / f"{env}.overlay{base.suffix}",
|
||||
]
|
||||
for c in candidates:
|
||||
if c.exists():
|
||||
overlay = c
|
||||
break
|
||||
|
||||
return base, overlay
|
||||
|
||||
|
||||
def load_config(base_path: str, env: str = None) -> dict:
|
||||
"""Load base config and merge with environment overlay.
|
||||
|
||||
Args:
|
||||
base_path: Path to base config YAML
|
||||
env: Environment name (dev, prod, cron, gateway, or custom)
|
||||
|
||||
Returns:
|
||||
Merged config dict
|
||||
"""
|
||||
base_file, overlay_file = find_config(base_path, env)
|
||||
|
||||
with open(base_file) as f:
|
||||
base_config = yaml.safe_load(f) or {}
|
||||
|
||||
if overlay_file and overlay_file.exists():
|
||||
with open(overlay_file) as f:
|
||||
overlay_config = yaml.safe_load(f) or {}
|
||||
return deep_merge(base_config, overlay_config)
|
||||
|
||||
return base_config
|
||||
|
||||
|
||||
def detect_env() -> str:
|
||||
"""Auto-detect environment from env vars.
|
||||
|
||||
Checks: TIMMY_ENV, HERMES_ENV, ENVIRONMENT, NODE_ENV
|
||||
"""
|
||||
for var in ('TIMMY_ENV', 'HERMES_ENV', 'ENVIRONMENT', 'NODE_ENV'):
|
||||
val = os.environ.get(var, '').strip().lower()
|
||||
if val:
|
||||
return val
|
||||
return None
|
||||
|
||||
|
||||
def load_config_auto(base_path: str) -> dict:
|
||||
"""Load config with auto-detected environment."""
|
||||
env = detect_env()
|
||||
return load_config(base_path, env=env)
|
||||
|
||||
|
||||
def list_overlays(base_path: str) -> list:
|
||||
"""List available overlay files for a base config."""
|
||||
base = Path(base_path)
|
||||
pattern = f"{base.stem}.*{base.suffix}"
|
||||
overlays = []
|
||||
for f in sorted(base.parent.glob(pattern)):
|
||||
if f == base:
|
||||
continue
|
||||
env_name = f.stem.replace(base.stem + '.', '')
|
||||
overlays.append({
|
||||
'env': env_name,
|
||||
'path': str(f),
|
||||
})
|
||||
return overlays
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import sys
|
||||
import json
|
||||
|
||||
base = sys.argv[1] if len(sys.argv) > 1 else 'config.yaml'
|
||||
env = sys.argv[2] if len(sys.argv) > 2 else None
|
||||
|
||||
config = load_config(base, env=env)
|
||||
print(json.dumps(config, indent=2, default=str))
|
||||
Reference in New Issue
Block a user