212 lines
7.3 KiB
Python
212 lines
7.3 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Config Template System — Environment-Specific Overlays (Issue #696)
|
||
|
|
|
||
|
|
Loads base.yaml + {env}.overlay.yaml with deep merge.
|
||
|
|
Overlay keys override base keys. Supports dot notation access.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
from scripts.config_template import ConfigTemplate, load_config
|
||
|
|
|
||
|
|
config = load_config("dev")
|
||
|
|
template = ConfigTemplate()
|
||
|
|
template.load("prod")
|
||
|
|
model = template.get("model.name")
|
||
|
|
|
||
|
|
CLI:
|
||
|
|
python3 scripts/config_template.py --env prod
|
||
|
|
python3 scripts/config_template.py --env dev --diff
|
||
|
|
python3 scripts/config_template.py --env prod --validate
|
||
|
|
python3 scripts/config_template.py --list-envs
|
||
|
|
"""
|
||
|
|
import argparse
|
||
|
|
import copy
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
try:
|
||
|
|
import yaml
|
||
|
|
except ImportError:
|
||
|
|
yaml = None
|
||
|
|
|
||
|
|
|
||
|
|
CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
|
||
|
|
KNOWN_ENVS = ("dev", "prod", "cron", "gateway")
|
||
|
|
|
||
|
|
|
||
|
|
def _deep_merge(base: dict, overlay: dict) -> dict:
|
||
|
|
"""Deep merge overlay into base. Overlay values win on conflict."""
|
||
|
|
result = copy.deepcopy(base)
|
||
|
|
for key, value in overlay.items():
|
||
|
|
if 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 _get_dotted(data: dict, key: str, default: Any = None) -> Any:
|
||
|
|
"""Get value from dict using dot notation: 'model.name' -> data['model']['name']."""
|
||
|
|
parts = key.split(".")
|
||
|
|
current = data
|
||
|
|
for part in parts:
|
||
|
|
if isinstance(current, dict) and part in current:
|
||
|
|
current = current[part]
|
||
|
|
else:
|
||
|
|
return default
|
||
|
|
return current
|
||
|
|
|
||
|
|
|
||
|
|
def _diff_dicts(base: dict, overlay: dict, prefix: str = "") -> List[dict]:
|
||
|
|
"""Compute diff between base and overlay configs."""
|
||
|
|
diffs = []
|
||
|
|
all_keys = set(list(base.keys()) + list(overlay.keys()))
|
||
|
|
for key in sorted(all_keys):
|
||
|
|
path = f"{prefix}.{key}" if prefix else key
|
||
|
|
in_base = key in base
|
||
|
|
in_overlay = key in overlay
|
||
|
|
if in_base and not in_overlay:
|
||
|
|
diffs.append({"key": path, "type": "removed_in_overlay", "base": base[key]})
|
||
|
|
elif not in_base and in_overlay:
|
||
|
|
diffs.append({"key": path, "type": "added_in_overlay", "overlay": overlay[key]})
|
||
|
|
elif isinstance(base[key], dict) and isinstance(overlay[key], dict):
|
||
|
|
diffs.extend(_diff_dicts(base[key], overlay[key], path))
|
||
|
|
elif base[key] != overlay[key]:
|
||
|
|
diffs.append({
|
||
|
|
"key": path, "type": "changed",
|
||
|
|
"base": base[key], "overlay": overlay[key]
|
||
|
|
})
|
||
|
|
return diffs
|
||
|
|
|
||
|
|
|
||
|
|
def _validate_config(config: dict) -> List[str]:
|
||
|
|
"""Validate config structure, return list of warnings."""
|
||
|
|
warnings = []
|
||
|
|
if "model" not in config:
|
||
|
|
warnings.append("Missing 'model' section")
|
||
|
|
elif "name" not in config.get("model", {}):
|
||
|
|
warnings.append("Missing 'model.name'")
|
||
|
|
if "provider" not in config:
|
||
|
|
warnings.append("Missing 'provider' section")
|
||
|
|
for key in config:
|
||
|
|
if not isinstance(key, str):
|
||
|
|
warnings.append(f"Non-string key: {key!r}")
|
||
|
|
return warnings
|
||
|
|
|
||
|
|
|
||
|
|
def _load_yaml_file(path: Path) -> dict:
|
||
|
|
"""Load a YAML file, return empty dict if missing."""
|
||
|
|
if not path.exists():
|
||
|
|
return {}
|
||
|
|
if yaml is None:
|
||
|
|
raise ImportError("PyYAML required: pip install pyyaml")
|
||
|
|
with open(path) as f:
|
||
|
|
data = yaml.safe_load(f)
|
||
|
|
return data if isinstance(data, dict) else {}
|
||
|
|
|
||
|
|
|
||
|
|
class ConfigTemplate:
|
||
|
|
"""Environment-specific config template with overlay merge."""
|
||
|
|
|
||
|
|
def __init__(self, config_dir: Optional[str] = None):
|
||
|
|
self.config_dir = Path(config_dir) if config_dir else CONFIG_DIR
|
||
|
|
self.base: Dict[str, Any] = {}
|
||
|
|
self.overlay: Dict[str, Any] = {}
|
||
|
|
self.merged: Dict[str, Any] = {}
|
||
|
|
self.env: Optional[str] = None
|
||
|
|
|
||
|
|
def load(self, env: str) -> dict:
|
||
|
|
"""Load base + overlay for the given environment."""
|
||
|
|
if env not in KNOWN_ENVS:
|
||
|
|
raise ValueError(f"Unknown environment '{env}'. Known: {', '.join(KNOWN_ENVS)}")
|
||
|
|
self.env = env
|
||
|
|
self.base = _load_yaml_file(self.config_dir / "base.yaml")
|
||
|
|
self.overlay = _load_yaml_file(self.config_dir / f"{env}.overlay.yaml")
|
||
|
|
self.merged = _deep_merge(self.base, self.overlay)
|
||
|
|
return self.merged
|
||
|
|
|
||
|
|
def get(self, key: str, default: Any = None) -> Any:
|
||
|
|
"""Get value with dot notation from merged config."""
|
||
|
|
return _get_dotted(self.merged, key, default)
|
||
|
|
|
||
|
|
def diff(self) -> List[dict]:
|
||
|
|
"""Show diff between base and current overlay."""
|
||
|
|
return _diff_dicts(self.base, self.overlay)
|
||
|
|
|
||
|
|
def validate(self) -> List[str]:
|
||
|
|
"""Validate merged config structure."""
|
||
|
|
return _validate_config(self.merged)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def list_environments() -> List[str]:
|
||
|
|
"""List known environments."""
|
||
|
|
return list(KNOWN_ENVS)
|
||
|
|
|
||
|
|
|
||
|
|
def load_config(env: str, config_dir: Optional[str] = None) -> dict:
|
||
|
|
"""One-shot: load merged config for an environment."""
|
||
|
|
t = ConfigTemplate(config_dir)
|
||
|
|
return t.load(env)
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description="Config Template System")
|
||
|
|
parser.add_argument("--env", default="dev", help="Environment name")
|
||
|
|
parser.add_argument("--diff", action="store_true", help="Show diff between base and overlay")
|
||
|
|
parser.add_argument("--validate", action="store_true", help="Validate merged config")
|
||
|
|
parser.add_argument("--list-envs", action="store_true", help="List known environments")
|
||
|
|
parser.add_argument("--config-dir", default=None, help="Config directory path")
|
||
|
|
parser.add_argument("--json", action="store_true", help="Output as JSON")
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
if args.list_envs:
|
||
|
|
envs = ConfigTemplate.list_environments()
|
||
|
|
if args.json:
|
||
|
|
print(json.dumps(envs, indent=2))
|
||
|
|
else:
|
||
|
|
for e in envs:
|
||
|
|
print(f" {e}")
|
||
|
|
return
|
||
|
|
|
||
|
|
template = ConfigTemplate(args.config_dir)
|
||
|
|
template.load(args.env)
|
||
|
|
|
||
|
|
if args.diff:
|
||
|
|
diffs = template.diff()
|
||
|
|
if args.json:
|
||
|
|
print(json.dumps(diffs, indent=2))
|
||
|
|
else:
|
||
|
|
if not diffs:
|
||
|
|
print(f"No differences between base and {args.env} overlay")
|
||
|
|
for d in diffs:
|
||
|
|
if d["type"] == "changed":
|
||
|
|
print(f" {d['key']}: {d['base']!r} -> {d['overlay']!r}")
|
||
|
|
elif d["type"] == "added_in_overlay":
|
||
|
|
print(f" {d['key']}: + {d['overlay']!r}")
|
||
|
|
elif d["type"] == "removed_in_overlay":
|
||
|
|
print(f" {d['key']}: - {d['base']!r}")
|
||
|
|
elif args.validate:
|
||
|
|
warnings = template.validate()
|
||
|
|
if args.json:
|
||
|
|
print(json.dumps({"valid": len(warnings) == 0, "warnings": warnings}, indent=2))
|
||
|
|
else:
|
||
|
|
if warnings:
|
||
|
|
for w in warnings:
|
||
|
|
print(f" WARNING: {w}")
|
||
|
|
else:
|
||
|
|
print(f" Config valid for {args.env}")
|
||
|
|
else:
|
||
|
|
if args.json:
|
||
|
|
print(json.dumps(template.merged, indent=2))
|
||
|
|
else:
|
||
|
|
print(f"Config for {args.env}:")
|
||
|
|
for k, v in template.merged.items():
|
||
|
|
print(f" {k}: {v!r}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|