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

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 logging

19import os

20from collections.abc import Callable

21from dataclasses import MISSING, dataclass, fields, is_dataclass

22from enum import Enum

23from pathlib import Path

24from typing import (

25 Any,

26 TypeVar,

27 Union,

28 get_args,

29 get_origin,

30 get_type_hints,

31)

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

35T = TypeVar("T")

37# ---------------------------------------------------------------------------

38# Exceptions

39# ---------------------------------------------------------------------------

42class ConfigError(Exception):

43 """Base configuration error."""

46class ConfigValidationError(ConfigError):

47 """Configuration value validation failure."""

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

50 self.field_path = field_path

51 self.value = value

52 self.reason = reason

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

56class ConfigNotFoundError(ConfigError):

57 """Required configuration key not found."""

60class ConfigSourceError(ConfigError):

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

64# ---------------------------------------------------------------------------

65# Source Types

66# ---------------------------------------------------------------------------

69class SourceType(Enum):

70 ENV = "env"

71 DOTENV = "dotenv"

72 YAML = "yaml"

73 TOML = "toml"

74 JSON = "json"

75 DICT = "dict"

78@dataclass

79class ConfigSource:

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

82 source_type: SourceType

83 data: dict[str, Any]

84 precedence: int = 100

85 description: str = ""

87 @classmethod

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

89 data: dict[str, Any] = {}

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

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

92 continue

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

94 # parse simple types

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

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

98 @classmethod

99 def from_dotenv(

100 cls, path: str | Path, precedence: int = 300, override: bool = False

101 ) -> ConfigSource:

102 from pathlib import Path

103

104 path = Path(path)

105 if not path.exists():

106 if override:

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

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

109 data: dict[str, Any] = {}

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

111 line = line.strip()

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

113 continue

114 if "=" not in line:

115 continue

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

117 key = key.strip().lower()

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

119 data[key] = _parse_env_value(val)

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

121

122 @classmethod

123 def from_yaml(cls, path: str | Path, precedence: int = 400) -> ConfigSource:

124 import yaml

125

126 path = Path(path)

127 if not path.exists():

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

129 with open(path) as f:

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

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

132

133 @classmethod

134 def from_toml(cls, path: str | Path, precedence: int = 400) -> ConfigSource:

135 path = Path(path)

136 if not path.exists():

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

138 try:

139 import tomllib

140 except ImportError:

141 import tomli as tomllib # type: ignore

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

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

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

145

146 @classmethod

147 def from_json(cls, path: str | Path, precedence: int = 400) -> ConfigSource:

148 path = Path(path)

149 if not path.exists():

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

151 with open(path) as f:

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

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

154

155 @classmethod

156 def from_dict(

157 cls, d: dict[str, Any], precedence: int = 500, description: str = "dict"

158 ) -> ConfigSource:

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

160

161

162# ---------------------------------------------------------------------------

163# Config Manager

164# ---------------------------------------------------------------------------

165

166

167class ConfigManager:

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

169

170 Usage:

171 cm = ConfigManager()

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

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

174

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

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

177

178 # Bind to dataclass

179 from dataclasses import dataclass

180

181 @dataclass

182 class AppConfig:

183 host: str = "0.0.0.0"

184 port: int = 8080

185 debug: bool = False

186

187 app_cfg = cm.bind(AppConfig)

188 """

189

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

191 self._sources: list[ConfigSource] = []

192 self._merged: dict[str, Any] | None = None

193 self._listeners: list[Callable[[str, Any, Any], None]] = []

194 self._secrets: set[str] = set()

195 if auto_env:

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

197

198 # -- Source management --

199

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

201 self._sources.append(source)

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

203 self._merged = None

204

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

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

207

208 def mark_secrets(self, keys: list[str]) -> None:

209 for k in keys:

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

211

212 # -- Merge --

213

214 def _merge_sources(self) -> dict[str, Any]:

215 result: dict[str, Any] = {}

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

217 for source in reversed(self._sources):

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

219 result[key] = val

220 return result

221

222 def _ensure_merged(self) -> dict[str, Any]:

223 if self._merged is None:

224 self._merged = self._merge_sources()

225 return self._merged

226

227 # -- Read --

228

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

230 merged = self._ensure_merged()

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

232 if value is MISSING:

233 if default is not MISSING:

234 return default

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

236 return value

237

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

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

240

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

242 val = self.get(key, default)

243 return int(val)

244

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

246 val = self.get(key, default)

247 return float(val)

248

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

250 val = self.get(key, default)

251 if isinstance(val, bool):

252 return val

253 if isinstance(val, str):

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

255 return bool(val)

256

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

258 val = self.get(key, default)

259 if isinstance(val, list):

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

261 if isinstance(val, str):

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

263 return [str(val)]

264

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

266 val = self.get(key, default)

267 if isinstance(val, dict):

268 return val

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

270

271 def keys(self) -> list[str]:

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

273

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

275 d = dict(self._ensure_merged())

276 if mask_secrets:

277 for sk in self._secrets:

278 if sk in d:

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

280 return d

281

282 # -- Bind to dataclass --

283

284 def bind(self, cls: type[T]) -> T:

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

286 hints = get_type_hints(cls)

287 kwargs: dict[str, Any] = {}

288 for fld in fields(cls):

289 key = fld.name.lower()

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

291 try:

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

293 except Exception:

294 raw = MISSING

295 if raw is MISSING:

296 if fld.default is not MISSING:

297 kwargs[fld.name] = fld.default

298 elif fld.default_factory is not MISSING:

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

300 else:

301 raise ConfigNotFoundError(

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

303 )

304 else:

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

306 return cls(**kwargs)

307

308 # -- Listener --

309

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

311 self._listeners.append(callback)

312

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

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

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

316 for cb in self._listeners:

317 cb(key, old, value)

318

319 def reload(self) -> None:

320 self._merged = None

321 self._ensure_merged()

322

323 def __repr__(self) -> str:

324 keys = self.keys()

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

326

327

328# ---------------------------------------------------------------------------

329# Helpers

330# ---------------------------------------------------------------------------

331

332

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

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

335 v = val.strip()

336 # bool

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

338 return True

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

340 return False

341 # null

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

343 return None

344 # int

345 try:

346 return int(v)

347 except ValueError:

348 pass

349 # float

350 try:

351 return float(v)

352 except ValueError:

353 pass

354 # JSON

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

356 try:

357 return json.loads(v)

358 except (json.JSONDecodeError, ValueError):

359 pass

360 return v

361

362

363def _flatten_dict(d: dict[str, Any], parent_key: str = "", sep: str = "_") -> dict[str, Any]:

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

365 items: list[tuple[str, Any]] = []

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

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

368 if (

369 isinstance(v, dict)

370 and not any(isinstance(v, t) for t in (list, tuple, set))

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

372 ):

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

374 else:

375 items.append((new_key, v))

376 return dict(items)

377

378

379def _coerce(value: Any, target_type: type) -> Any:

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

381 if value is None:

382 return None

383 origin = get_origin(target_type)

384 if origin is Union:

385 args = get_args(target_type)

386 if type(None) in args:

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

388 if value is None:

389 return None

390 if non_none:

391 return _coerce(value, non_none[0])

392 if target_type is bool:

393 if isinstance(value, bool):

394 return value

395 if isinstance(value, str):

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

397 return bool(value)

398 if target_type is int:

399 return int(value)

400 if target_type is float:

401 return float(value)

402 if target_type is str:

403 return str(value)

404 if target_type is list or origin is list:

405 if isinstance(value, list):

406 return value

407 if isinstance(value, str):

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

409 return [value]

410 if origin is dict:

411 if isinstance(value, dict):

412 return value

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

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

415 return target_type(**value)

416 return value