Coverage for agentos/core/config.py: 69%

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

13"""

15from __future__ import annotations

17import json

18import os

19from dataclasses import dataclass, fields, is_dataclass, MISSING

20from enum import Enum

21from pathlib import Path

22from typing import (

23 Any, Callable, Dict, List, Optional, Set, Tuple, Type, TypeVar, Union,

24 get_args, get_origin, get_type_hints,

25)

26import logging

28logger = logging.getLogger("agentos.config")

30T = TypeVar("T")

32# ---------------------------------------------------------------------------

33# Exceptions

34# ---------------------------------------------------------------------------

36class ConfigError(Exception):

37 """Base configuration error."""

40class ConfigValidationError(ConfigError):

41 """Configuration value validation failure."""

43 def __init__(self, field_path: str, value: Any, reason: str):

44 self.field_path = field_path

45 self.value = value

46 self.reason = reason

47 super().__init__(f"{field_path}: {reason} (got {value!r})")

50class ConfigNotFoundError(ConfigError):

51 """Required configuration key not found."""

54class ConfigSourceError(ConfigError):

55 """Failed to load configuration from a source."""

58# ---------------------------------------------------------------------------

59# Source Types

60# ---------------------------------------------------------------------------

62class SourceType(Enum):

63 ENV = "env"

64 DOTENV = "dotenv"

65 YAML = "yaml"

66 TOML = "toml"

67 JSON = "json"

68 DICT = "dict"

71@dataclass

72class ConfigSource:

73 """Configuration source with precedence (lower number = higher priority)."""

74 source_type: SourceType

75 data: Dict[str, Any]

76 precedence: int = 100

77 description: str = ""

79 @classmethod

80 def from_env(cls, prefix: str = "", precedence: int = 200) -> "ConfigSource":

81 data: Dict[str, Any] = {}

82 for key, val in os.environ.items():

83 if prefix and not key.startswith(prefix):

84 continue

85 clean_key = key[len(prefix):] if prefix else key

86 # parse simple types

87 data[clean_key.lower()] = _parse_env_value(val)

88 return cls(SourceType.ENV, data, precedence, f"env(prefix={prefix!r})")

90 @classmethod

91 def from_dotenv(cls, path: Union[str, Path], precedence: int = 300,

92 override: bool = False) -> "ConfigSource":

93 from pathlib import Path as P

94 path = P(path)

95 if not path.exists():

96 if override:

97 return cls(SourceType.DOTENV, {}, precedence, f"dotenv({path})")

98 raise ConfigSourceError(f".env file not found: {path}")

99 data: Dict[str, Any] = {}

100 for line in path.read_text().splitlines():

101 line = line.strip()

102 if not line or line.startswith("#"):

103 continue

104 if "=" not in line:

105 continue

106 key, _, val = line.partition("=")

107 key = key.strip().lower()

108 val = val.strip().strip('"').strip("'")

109 data[key] = _parse_env_value(val)

110 return cls(SourceType.DOTENV, data, precedence, f"dotenv({path})")

111

112 @classmethod

113 def from_yaml(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":

114 import yaml

115 path = Path(path)

116 if not path.exists():

117 raise ConfigSourceError(f"YAML file not found: {path}")

118 with open(path) as f:

119 data = _flatten_dict(yaml.safe_load(f) or {})

120 return cls(SourceType.YAML, data, precedence, f"yaml({path})")

121

122 @classmethod

123 def from_toml(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":

124 path = Path(path)

125 if not path.exists():

126 raise ConfigSourceError(f"TOML file not found: {path}")

127 try:

128 import tomllib

129 except ImportError:

130 import tomli as tomllib # type: ignore

131 with open(path, "rb") as f:

132 data = _flatten_dict(tomllib.load(f) or {})

133 return cls(SourceType.TOML, data, precedence, f"toml({path})")

134

135 @classmethod

136 def from_json(cls, path: Union[str, Path], precedence: int = 400) -> "ConfigSource":

137 path = Path(path)

138 if not path.exists():

139 raise ConfigSourceError(f"JSON file not found: {path}")

140 with open(path) as f:

141 data = _flatten_dict(json.load(f) or {})

142 return cls(SourceType.JSON, data, precedence, f"json({path})")

143

144 @classmethod

145 def from_dict(cls, d: Dict[str, Any], precedence: int = 500,

146 description: str = "dict") -> "ConfigSource":

147 return cls(SourceType.DICT, dict(d), precedence, description)

148

149

150# ---------------------------------------------------------------------------

151# Config Manager

152# ---------------------------------------------------------------------------

153

154class ConfigManager:

155 """Central configuration manager with source merging and typed access.

156

157 Usage:

158 cm = ConfigManager()

159 cm.add_source(ConfigSource.from_env("AGENTOS_"))

160 cm.add_source(ConfigSource.from_yaml("config/prod.yaml"))

161

162 db_host = cm.get("database.host", default="localhost")

163 db_port = cm.get_int("database.port", default=5432)

164

165 # Bind to dataclass

166 from dataclasses import dataclass

167

168 @dataclass

169 class AppConfig:

170 host: str = "0.0.0.0"

171 port: int = 8080

172 debug: bool = False

173

174 app_cfg = cm.bind(AppConfig)

175 """

176

177 def __init__(self, auto_env: bool = True, env_prefix: str = "AGENTOS_"):

178 self._sources: List[ConfigSource] = []

179 self._merged: Optional[Dict[str, Any]] = None

180 self._listeners: List[Callable[[str, Any, Any], None]] = []

181 self._secrets: Set[str] = set()

182 if auto_env:

183 self.add_source(ConfigSource.from_env(env_prefix))

184

185 # -- Source management --

186

187 def add_source(self, source: ConfigSource) -> None:

188 self._sources.append(source)

189 self._sources.sort(key=lambda s: s.precedence)

190 self._merged = None

191

192 def mark_secret(self, key: str) -> None:

193 self._secrets.add(key.lower())

194

195 def mark_secrets(self, keys: List[str]) -> None:

196 for k in keys:

197 self._secrets.add(k.lower())

198

199 # -- Merge --

200

201 def _merge_sources(self) -> Dict[str, Any]:

202 result: Dict[str, Any] = {}

203 # Lower precedence = higher priority → iterate reversed so high-priority overrides

204 for source in reversed(self._sources):

205 for key, val in source.data.items():

206 result[key] = val

207 return result

208

209 def _ensure_merged(self) -> Dict[str, Any]:

210 if self._merged is None:

211 self._merged = self._merge_sources()

212 return self._merged

213

214 # -- Read --

215

216 def get(self, key: str, default: Any = MISSING) -> Any:

217 merged = self._ensure_merged()

218 value = merged.get(key.lower(), MISSING)

219 if value is MISSING:

220 if default is not MISSING:

221 return default

222 raise ConfigNotFoundError(f"Configuration key not found: {key}")

223 return value

224

225 def get_str(self, key: str, default: Any = MISSING) -> str:

226 return str(self.get(key, default))

227

228 def get_int(self, key: str, default: Any = MISSING) -> int:

229 val = self.get(key, default)

230 return int(val)

231

232 def get_float(self, key: str, default: Any = MISSING) -> float:

233 val = self.get(key, default)

234 return float(val)

235

236 def get_bool(self, key: str, default: Any = MISSING) -> bool:

237 val = self.get(key, default)

238 if isinstance(val, bool):

239 return val

240 if isinstance(val, str):

241 return val.lower() in ("true", "1", "yes", "on")

242 return bool(val)

243

244 def get_list(self, key: str, default: Any = MISSING, separator: str = ",") -> List[str]:

245 val = self.get(key, default)

246 if isinstance(val, list):

247 return [str(v) for v in val]

248 if isinstance(val, str):

249 return [v.strip() for v in val.split(separator) if v.strip()]

250 return [str(val)]

251

252 def get_dict(self, key: str, default: Any = MISSING) -> Dict[str, Any]:

253 val = self.get(key, default)

254 if isinstance(val, dict):

255 return val

256 raise ConfigValidationError(key, val, "expected dict")

257

258 def keys(self) -> List[str]:

259 return sorted(self._ensure_merged().keys())

260

261 def to_dict(self, mask_secrets: bool = True) -> Dict[str, Any]:

262 d = dict(self._ensure_merged())

263 if mask_secrets:

264 for sk in self._secrets:

265 if sk in d:

266 d[sk] = "***MASKED***"

267 return d

268

269 # -- Bind to dataclass --

270

271 def bind(self, cls: Type[T]) -> T:

272 """Bind merged configuration to a dataclass instance."""

273 hints = get_type_hints(cls)

274 kwargs: Dict[str, Any] = {}

275 for fld in fields(cls):

276 key = fld.name.lower()

277 field_type = hints.get(fld.name, str)

278 try:

279 raw = self._ensure_merged().get(key, MISSING)

280 except Exception:

281 raw = MISSING

282 if raw is MISSING:

283 if fld.default is not MISSING:

284 kwargs[fld.name] = fld.default

285 elif fld.default_factory is not MISSING:

286 kwargs[fld.name] = fld.default_factory()

287 else:

288 raise ConfigNotFoundError(

289 f"Required config '{fld.name}' not found and has no default"

290 )

291 else:

292 kwargs[fld.name] = _coerce(raw, field_type)

293 return cls(**kwargs)

294

295 # -- Listener --

296

297 def on_change(self, callback: Callable[[str, Any, Any], None]) -> None:

298 self._listeners.append(callback)

299

300 def set(self, key: str, value: Any) -> None:

301 old = self._ensure_merged().get(key.lower())

302 self._ensure_merged()[key.lower()] = value

303 for cb in self._listeners:

304 cb(key, old, value)

305

306 def reload(self) -> None:

307 self._merged = None

308 self._ensure_merged()

309

310 def __repr__(self) -> str:

311 keys = self.keys()

312 return f"ConfigManager(sources={len(self._sources)}, keys={len(keys)})"

313

314

315# ---------------------------------------------------------------------------

316# Helpers

317# ---------------------------------------------------------------------------

318

319def _parse_env_value(val: str) -> Any:

320 """Parse string environment value to native types."""

321 v = val.strip()

322 # bool

323 if v.lower() in ("true", "yes", "on"):

324 return True

325 if v.lower() in ("false", "no", "off"):

326 return False

327 # null

328 if v.lower() in ("null", "none", ""):

329 return None

330 # int

331 try:

332 return int(v)

333 except ValueError:

334 pass

335 # float

336 try:

337 return float(v)

338 except ValueError:

339 pass

340 # JSON

341 if (v.startswith("{") and v.endswith("}")) or (v.startswith("[") and v.endswith("]")):

342 try:

343 return json.loads(v)

344 except (json.JSONDecodeError, ValueError):

345 pass

346 return v

347

348

349def _flatten_dict(d: Dict[str, Any], parent_key: str = "",

350 sep: str = "_") -> Dict[str, Any]:

351 """Flatten nested dicts into dot-separated keys."""

352 items: List[Tuple[str, Any]] = []

353 for k, v in d.items():

354 new_key = f"{parent_key}{sep}{k}".lower() if parent_key else k.lower()

355 if isinstance(v, dict) and not any(

356 isinstance(v, t) for t in (list, tuple, set)

357 ) and not (k.isupper() and all(c.isupper() or c == "_" for c in k)):

358 items.extend(_flatten_dict(v, new_key, sep).items())

359 else:

360 items.append((new_key, v))

361 return dict(items)

362

363

364def _coerce(value: Any, target_type: Type) -> Any:

365 """Coerce a value to the target type."""

366 if value is None:

367 return None

368 origin = get_origin(target_type)

369 if origin is Union:

370 args = get_args(target_type)

371 if type(None) in args:

372 non_none = [a for a in args if a is not type(None)]

373 if value is None:

374 return None

375 if non_none:

376 return _coerce(value, non_none[0])

377 if target_type is bool:

378 if isinstance(value, bool):

379 return value

380 if isinstance(value, str):

381 return value.lower() in ("true", "1", "yes", "on")

382 return bool(value)

383 if target_type is int:

384 return int(value)

385 if target_type is float:

386 return float(value)

387 if target_type is str:

388 return str(value)

389 if target_type is list or origin is list:

390 if isinstance(value, list):

391 return value

392 if isinstance(value, str):

393 return [v.strip() for v in value.split(",") if v.strip()]

394 return [value]

395 if origin is dict:

396 if isinstance(value, dict):

397 return value

398 raise ConfigValidationError("", value, "cannot coerce to dict")

399 if is_dataclass(target_type) and isinstance(value, dict):

400 return target_type(**value)

401 return value