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