Coverage for agentos/core/config.py: 0%
263 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 08:01 +0800
1"""
2Production-grade typed configuration management.
4Supports:
5- Multiple sources: env vars, .env files, YAML, TOML, JSON, dict
6- Strict typing with Pydantic-style validation
7- Hierarchical merging with precedence
8- Environment-specific overrides (dev/staging/prod)
9- Secret masking in logs
10- Hot-reload with callbacks
12Copyright 2026 AgentOS. All rights reserved.
13"""
15from __future__ import annotations
17import json
18import os
19from copy import deepcopy
20from dataclasses import dataclass, field, fields, is_dataclass, MISSING
21from enum import Enum
22from pathlib import Path
23from typing import (
24 Any, Callable, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union,
25 get_args, get_origin, get_type_hints, ClassVar,
26)
27import logging
29logger = logging.getLogger("agentos.config")
31T = TypeVar("T")
33# ---------------------------------------------------------------------------
34# Exceptions
35# ---------------------------------------------------------------------------
37class ConfigError(Exception):
38 """Base configuration error."""
41class ConfigValidationError(ConfigError):
42 """Configuration value validation failure."""
44 def __init__(self, field_path: str, value: Any, reason: str):
45 self.field_path = field_path
46 self.value = value
47 self.reason = reason
48 super().__init__(f"{field_path}: {reason} (got {value!r})")
51class ConfigNotFoundError(ConfigError):
52 """Required configuration key not found."""
55class ConfigSourceError(ConfigError):
56 """Failed to load configuration from a source."""
59# ---------------------------------------------------------------------------
60# Source Types
61# ---------------------------------------------------------------------------
63class SourceType(Enum):
64 ENV = "env"
65 DOTENV = "dotenv"
66 YAML = "yaml"
67 TOML = "toml"
68 JSON = "json"
69 DICT = "dict"
72@dataclass
73class ConfigSource:
74 """Configuration source with precedence (lower number = higher priority)."""
75 source_type: SourceType
76 data: Dict[str, Any]
77 precedence: int = 100
78 description: str = ""
80 @classmethod
81 def from_env(cls, prefix: str = "", precedence: int = 200) -> "ConfigSource":
82 data: Dict[str, Any] = {}
83 for key, val in os.environ.items():
84 if prefix and not key.startswith(prefix):
85 continue
86 clean_key = key[len(prefix):] if prefix else key
87 # parse simple types
88 data[clean_key.lower()] = _parse_env_value(val)
89 return cls(SourceType.ENV, data, precedence, f"env(prefix={prefix!r})")
91 @classmethod
92 def from_dotenv(cls, path: Union[str, Path], precedence: int = 300,
93 override: bool = False) -> "ConfigSource":
94 from pathlib import Path as P
95 path = P(path)
96 if not path.exists():
97 if override:
98 return cls(SourceType.DOTENV, {}, precedence, f"dotenv({path})")
99 raise ConfigSourceError(f".env file not found: {path}")
100 data: Dict[str, Any] = {}
101 for line in path.read_text().splitlines():
102 line = line.strip()
103 if not line or line.startswith("#"):
104 continue
105 if "=" not in line:
106 continue
107 key, _, val = line.partition("=")
108 key = key.strip().lower()
109 val = val.strip().strip('"').strip("'")
110 data[key] = _parse_env_value(val)
111 return cls(SourceType.DOTENV, data, precedence, f"dotenv({path})")
113 @classmethod
114 def from_yaml(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":
115 import yaml
116 path = Path(path)
117 if not path.exists():
118 raise ConfigSourceError(f"YAML file not found: {path}")
119 with open(path) as f:
120 data = _flatten_dict(yaml.safe_load(f) or {})
121 return cls(SourceType.YAML, data, precedence, f"yaml({path})")
123 @classmethod
124 def from_toml(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":
125 path = Path(path)
126 if not path.exists():
127 raise ConfigSourceError(f"TOML file not found: {path}")
128 try:
129 import tomllib
130 except ImportError:
131 import tomli as tomllib # type: ignore
132 with open(path, "rb") as f:
133 data = _flatten_dict(tomllib.load(f) or {})
134 return cls(SourceType.TOML, data, precedence, f"toml({path})")
136 @classmethod
137 def from_json(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":
138 path = Path(path)
139 if not path.exists():
140 raise ConfigSourceError(f"JSON file not found: {path}")
141 with open(path) as f:
142 data = _flatten_dict(json.load(f) or {})
143 return cls(SourceType.JSON, data, precedence, f"json({path})")
145 @classmethod
146 def from_dict(cls, d: Dict[str, Any], precedence: int = 500,
147 description: str = "dict") -> "ConfigSource":
148 return cls(SourceType.DICT, dict(d), precedence, description)
151# ---------------------------------------------------------------------------
152# Config Manager
153# ---------------------------------------------------------------------------
155class ConfigManager:
156 """Central configuration manager with source merging and typed access.
158 Usage:
159 cm = ConfigManager()
160 cm.add_source(ConfigSource.from_env("AGENTOS_"))
161 cm.add_source(ConfigSource.from_yaml("config/prod.yaml"))
163 db_host = cm.get("database.host", default="localhost")
164 db_port = cm.get_int("database.port", default=5432)
166 # Bind to dataclass
167 from dataclasses import dataclass
169 @dataclass
170 class AppConfig:
171 host: str = "0.0.0.0"
172 port: int = 8080
173 debug: bool = False
175 app_cfg = cm.bind(AppConfig)
176 """
178 def __init__(self, auto_env: bool = True, env_prefix: str = "AGENTOS_"):
179 self._sources: List[ConfigSource] = []
180 self._merged: Optional[Dict[str, Any]] = None
181 self._listeners: List[Callable[[str, Any, Any], None]] = []
182 self._secrets: Set[str] = set()
183 if auto_env:
184 self.add_source(ConfigSource.from_env(env_prefix))
186 # -- Source management --
188 def add_source(self, source: ConfigSource) -> None:
189 self._sources.append(source)
190 self._sources.sort(key=lambda s: s.precedence)
191 self._merged = None
193 def mark_secret(self, key: str) -> None:
194 self._secrets.add(key.lower())
196 def mark_secrets(self, keys: List[str]) -> None:
197 for k in keys:
198 self._secrets.add(k.lower())
200 # -- Merge --
202 def _merge_sources(self) -> Dict[str, Any]:
203 result: Dict[str, Any] = {}
204 # Lower precedence = higher priority → iterate reversed so high-priority overrides
205 for source in reversed(self._sources):
206 for key, val in source.data.items():
207 result[key] = val
208 return result
210 def _ensure_merged(self) -> Dict[str, Any]:
211 if self._merged is None:
212 self._merged = self._merge_sources()
213 return self._merged
215 # -- Read --
217 def get(self, key: str, default: Any = MISSING) -> Any:
218 merged = self._ensure_merged()
219 value = merged.get(key.lower(), MISSING)
220 if value is MISSING:
221 if default is not MISSING:
222 return default
223 raise ConfigNotFoundError(f"Configuration key not found: {key}")
224 return value
226 def get_str(self, key: str, default: Any = MISSING) -> str:
227 return str(self.get(key, default))
229 def get_int(self, key: str, default: Any = MISSING) -> int:
230 val = self.get(key, default)
231 return int(val)
233 def get_float(self, key: str, default: Any = MISSING) -> float:
234 val = self.get(key, default)
235 return float(val)
237 def get_bool(self, key: str, default: Any = MISSING) -> bool:
238 val = self.get(key, default)
239 if isinstance(val, bool):
240 return val
241 if isinstance(val, str):
242 return val.lower() in ("true", "1", "yes", "on")
243 return bool(val)
245 def get_list(self, key: str, default: Any = MISSING, separator: str = ",") -> List[str]:
246 val = self.get(key, default)
247 if isinstance(val, list):
248 return [str(v) for v in val]
249 if isinstance(val, str):
250 return [v.strip() for v in val.split(separator) if v.strip()]
251 return [str(val)]
253 def get_dict(self, key: str, default: Any = MISSING) -> Dict[str, Any]:
254 val = self.get(key, default)
255 if isinstance(val, dict):
256 return val
257 raise ConfigValidationError(key, val, "expected dict")
259 def keys(self) -> List[str]:
260 return sorted(self._ensure_merged().keys())
262 def to_dict(self, mask_secrets: bool = True) -> Dict[str, Any]:
263 d = dict(self._ensure_merged())
264 if mask_secrets:
265 for sk in self._secrets:
266 if sk in d:
267 d[sk] = "***MASKED***"
268 return d
270 # -- Bind to dataclass --
272 def bind(self, cls: Type[T]) -> T:
273 """Bind merged configuration to a dataclass instance."""
274 hints = get_type_hints(cls)
275 kwargs: Dict[str, Any] = {}
276 for fld in fields(cls):
277 key = fld.name.lower()
278 field_type = hints.get(fld.name, str)
279 try:
280 raw = self._ensure_merged().get(key, MISSING)
281 except Exception:
282 raw = MISSING
283 if raw is MISSING:
284 if fld.default is not MISSING:
285 kwargs[fld.name] = fld.default
286 elif fld.default_factory is not MISSING:
287 kwargs[fld.name] = fld.default_factory()
288 else:
289 raise ConfigNotFoundError(
290 f"Required config '{fld.name}' not found and has no default"
291 )
292 else:
293 kwargs[fld.name] = _coerce(raw, field_type)
294 return cls(**kwargs)
296 # -- Listener --
298 def on_change(self, callback: Callable[[str, Any, Any], None]) -> None:
299 self._listeners.append(callback)
301 def set(self, key: str, value: Any) -> None:
302 old = self._ensure_merged().get(key.lower())
303 self._ensure_merged()[key.lower()] = value
304 for cb in self._listeners:
305 cb(key, old, value)
307 def reload(self) -> None:
308 self._merged = None
309 self._ensure_merged()
311 def __repr__(self) -> str:
312 keys = self.keys()
313 return f"ConfigManager(sources={len(self._sources)}, keys={len(keys)})"
316# ---------------------------------------------------------------------------
317# Helpers
318# ---------------------------------------------------------------------------
320def _parse_env_value(val: str) -> Any:
321 """Parse string environment value to native types."""
322 v = val.strip()
323 # bool
324 if v.lower() in ("true", "yes", "on"):
325 return True
326 if v.lower() in ("false", "no", "off"):
327 return False
328 # null
329 if v.lower() in ("null", "none", ""):
330 return None
331 # int
332 try:
333 return int(v)
334 except ValueError:
335 pass
336 # float
337 try:
338 return float(v)
339 except ValueError:
340 pass
341 # JSON
342 if (v.startswith("{") and v.endswith("}")) or (v.startswith("[") and v.endswith("]")):
343 try:
344 return json.loads(v)
345 except (json.JSONDecodeError, ValueError):
346 pass
347 return v
350def _flatten_dict(d: Dict[str, Any], parent_key: str = "",
351 sep: str = "_") -> Dict[str, Any]:
352 """Flatten nested dicts into dot-separated keys."""
353 items: List[Tuple[str, Any]] = []
354 for k, v in d.items():
355 new_key = f"{parent_key}{sep}{k}".lower() if parent_key else k.lower()
356 if isinstance(v, dict) and not any(
357 isinstance(v, t) for t in (list, tuple, set)
358 ) and not (k.isupper() and all(c.isupper() or c == "_" for c in k)):
359 items.extend(_flatten_dict(v, new_key, sep).items())
360 else:
361 items.append((new_key, v))
362 return dict(items)
365def _coerce(value: Any, target_type: Type) -> Any:
366 """Coerce a value to the target type."""
367 if value is None:
368 return None
369 origin = get_origin(target_type)
370 if origin is Union:
371 args = get_args(target_type)
372 if type(None) in args:
373 non_none = [a for a in args if a is not type(None)]
374 if value is None:
375 return None
376 if non_none:
377 return _coerce(value, non_none[0])
378 if target_type is bool:
379 if isinstance(value, bool):
380 return value
381 if isinstance(value, str):
382 return value.lower() in ("true", "1", "yes", "on")
383 return bool(value)
384 if target_type is int:
385 return int(value)
386 if target_type is float:
387 return float(value)
388 if target_type is str:
389 return str(value)
390 if target_type is list or origin is list:
391 if isinstance(value, list):
392 return value
393 if isinstance(value, str):
394 return [v.strip() for v in value.split(",") if v.strip()]
395 return [value]
396 if origin is dict:
397 if isinstance(value, dict):
398 return value
399 raise ConfigValidationError("", value, f"cannot coerce to dict")
400 if is_dataclass(target_type) and isinstance(value, dict):
401 return target_type(**value)
402 return value