Coverage for agentos/tools/config_manager.py: 92%
150 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 11:37 +0800
1"""
2ConfigManager — layered configuration with schema validation, env overlay, and hot reload.
4Layers (priority low → high):
5 1. defaults — hardcoded defaults
6 2. file — YAML/JSON config file(s)
7 3. env — environment variable overrides (PREFIX_KEY=value)
8 4. runtime — programmatic overrides via set()
10Supports: dot-path access, schema validation, file watching for hot reload.
11"""
13import json
14import os
15import threading
16from copy import deepcopy
17from pathlib import Path
18from typing import Any, Callable, Dict, List, Optional, Union
21# ============================================================================
22# Schema Validation
23# ============================================================================
25class ConfigSchemaError(Exception):
26 """Validation error with path and message."""
27 def __init__(self, path: str, message: str):
28 self.path = path
29 self.message = message
30 super().__init__(f"[{path}] {message}")
33class ConfigSchema:
34 """Declarative schema for config validation."""
36 def __init__(self):
37 self._fields: Dict[str, Dict[str, Any]] = {}
39 def field(
40 self,
41 name: str,
42 type_: type = str,
43 required: bool = False,
44 default: Any = None,
45 choices: Optional[List[Any]] = None,
46 min_val: Optional[float] = None,
47 max_val: Optional[float] = None,
48 description: str = "",
49 ) -> "ConfigSchema":
50 self._fields[name] = {
51 "type": type_,
52 "required": required,
53 "default": default,
54 "choices": choices,
55 "min": min_val,
56 "max": max_val,
57 "description": description,
58 }
59 return self
61 def validate(self, config: Dict[str, Any], prefix: str = "") -> List[ConfigSchemaError]:
62 errors = []
63 for name, spec in self._fields.items():
64 path = f"{prefix}.{name}" if prefix else name
65 value = config.get(name)
66 if value is None:
67 if spec["required"]:
68 errors.append(ConfigSchemaError(path, "required field missing"))
69 continue
70 if not isinstance(value, spec["type"]):
71 errors.append(ConfigSchemaError(path,
72 f"expected {spec['type'].__name__}, got {type(value).__name__}"))
73 continue
74 if spec["choices"] and value not in spec["choices"]:
75 errors.append(ConfigSchemaError(path,
76 f"invalid choice '{value}', allowed: {spec['choices']}"))
77 if spec["min"] is not None and value < spec["min"]:
78 errors.append(ConfigSchemaError(path, f"value {value} below min {spec['min']}"))
79 if spec["max"] is not None and value > spec["max"]:
80 errors.append(ConfigSchemaError(path, f"value {value} above max {spec['max']}"))
81 return errors
84# ============================================================================
85# ConfigManager
86# ============================================================================
88class ConfigManager:
89 """Layered configuration manager.
91 Usage:
92 cm = ConfigManager(defaults={"host": "localhost", "port": 8080})
93 cm.load_file("config.yaml")
94 cm.load_env("APP_") # APP_HOST=0.0.0.0 overrides host
95 cm.get("host") # returns "0.0.0.0"
96 """
98 def __init__(
99 self,
100 defaults: Optional[Dict[str, Any]] = None,
101 schema: Optional[ConfigSchema] = None,
102 ):
103 self._defaults = deepcopy(defaults) if defaults else {}
104 self._file_layer: Dict[str, Any] = {}
105 self._env_layer: Dict[str, Any] = {}
106 self._runtime_layer: Dict[str, Any] = {}
107 self._schema = schema
108 self._lock = threading.RLock()
109 self._watchers: Dict[str, float] = {} # path → mtime
110 self._on_change: List[Callable[[str, Any, Any], None]] = []
112 # ---------- file loading ----------
114 def load_file(self, path: Union[str, Path]) -> None:
115 """Load config from YAML or JSON file."""
116 path = Path(path)
117 if not path.exists():
118 raise FileNotFoundError(f"Config file not found: {path}")
119 content = path.read_text(encoding="utf-8")
120 if path.suffix in (".yaml", ".yml"):
121 data = self._parse_yaml(content)
122 else:
123 data = json.loads(content)
124 with self._lock:
125 self._file_layer = self._deep_merge(self._file_layer, data)
126 if str(path) not in self._watchers:
127 self._watchers[str(path)] = path.stat().st_mtime
129 def _parse_yaml(self, content: str) -> Dict[str, Any]:
130 try:
131 import yaml
132 return yaml.safe_load(content) or {}
133 except ImportError:
134 raise ImportError("PyYAML required for YAML config files. pip install pyyaml")
136 # ---------- env loading ----------
138 def load_env(self, prefix: str = "") -> None:
139 """Overlay environment variables. PREFIX_KEY → config key (lowercase)."""
140 with self._lock:
141 for key, value in os.environ.items():
142 if not prefix or key.startswith(prefix):
143 config_key = key[len(prefix):].lower() if prefix else key.lower()
144 # Try to parse numbers/booleans
145 parsed = self._parse_value(value)
146 self._env_layer[config_key] = parsed
148 @staticmethod
149 def _parse_value(value: str) -> Any:
150 if value.lower() in ("true", "false"):
151 return value.lower() == "true"
152 if value.lower() in ("null", "none", ""):
153 return None
154 try:
155 if "." in value:
156 return float(value)
157 return int(value)
158 except ValueError:
159 return value
161 # ---------- runtime access ----------
163 def set(self, key: str, value: Any) -> None:
164 old = self.get(key)
165 with self._lock:
166 self._runtime_layer[key] = value
167 new_val = value
168 if old != new_val:
169 self._notify(key, old, new_val)
171 def get(self, key: str, default: Any = None) -> Any:
172 with self._lock:
173 # Priority: runtime > env > file > defaults
174 if key in self._runtime_layer:
175 return self._runtime_layer[key]
176 if key in self._env_layer:
177 return self._env_layer[key]
178 if key in self._file_layer:
179 return self._file_layer[key]
180 if key in self._defaults:
181 return self._defaults[key]
182 return default
184 def get_dot(self, path: str, default: Any = None) -> Any:
185 """Dot-path access: get_dot('server.host') → get('server')['host']"""
186 keys = path.split(".")
187 current = None
188 for i, key in enumerate(keys):
189 if i == 0:
190 current = self.get(key)
191 elif isinstance(current, dict):
192 current = current.get(key)
193 else:
194 return default
195 if current is None:
196 return default
197 return current
199 def all(self) -> Dict[str, Any]:
200 """Return merged config dict."""
201 with self._lock:
202 result = deepcopy(self._defaults)
203 result = self._deep_merge(result, self._file_layer)
204 result = self._deep_merge(result, self._env_layer)
205 result = self._deep_merge(result, self._runtime_layer)
206 return result
208 def reload(self, force: bool = False) -> None:
209 """Reload file layer from disk (check mtime unless force=True)."""
210 with self._lock:
211 for path_str, mtime in list(self._watchers.items()):
212 p = Path(path_str)
213 if p.exists():
214 new_mtime = p.stat().st_mtime
215 if force or new_mtime > mtime:
216 self._file_layer = {}
217 self.load_file(path_str)
218 self._watchers[path_str] = new_mtime
220 # ---------- validation ----------
222 def validate(self) -> List[ConfigSchemaError]:
223 if not self._schema:
224 return []
225 return self._schema.validate(self.all())
227 # ---------- events ----------
229 def on_change(self, callback: Callable[[str, Any, Any], None]) -> None:
230 self._on_change.append(callback)
232 def _notify(self, key: str, old: Any, new: Any) -> None:
233 for cb in self._on_change:
234 try:
235 cb(key, old, new)
236 except Exception:
237 pass
239 # ---------- internal ----------
241 @staticmethod
242 def _deep_merge(base: Dict, overlay: Dict) -> Dict:
243 result = deepcopy(base)
244 for k, v in overlay.items():
245 if k in result and isinstance(result[k], dict) and isinstance(v, dict):
246 result[k] = ConfigManager._deep_merge(result[k], v)
247 else:
248 result[k] = v
249 return result