Coverage for src/lexigram/admin/settings/loader.py: 25%

1"""Configuration loader with layered sources. 

2 

3Loads configuration from: 

41. Pydantic defaults (AdminConfig) 

52. YAML file (application.yaml admin: section) 

63. Environment variables (LEX_ADMIN_*) 

74. Runtime overrides 

8 

9Supports hot-reload for runtime configuration updates. 

10All operations are async-only for consistency with the lexigram ecosystem. 

11""" 

12 

13from __future__ import annotations 

14 

15from collections.abc import Callable 

16from dataclasses import dataclass 

17import os 

18from pathlib import Path 

19from typing import Any, TypeVar 

20 

21import aiofiles 

22import yaml 

23 

24from lexigram import serialization as json 

25from lexigram.admin.config import AdminConfig 

26from lexigram.config import ConfigLoader as BaseConfigLoader 

27from lexigram.domain import DomainModel 

28 

29T = TypeVar("T") 

30 

31__all__ = ["AdminConfigLoader"] 

32 

33 

34@dataclass(init=False) 

35class ConfigPath(DomainModel): 

36 """Immutable path reference for nested config access.""" 

37 

38 parts: tuple[str, ...] 

39 

40 @classmethod 

41 def from_string(cls, path: str) -> ConfigPath: 

42 """Create from dot-separated string.""" 

43 return cls(parts=tuple(path.split("."))) 

44 

45 def __str__(self) -> str: 

46 return ".".join(self.parts) 

47 

48 

49class AdminConfigLoader: 

50 """Layered configuration loader for lexigram-admin. 

51 

52 Configuration priority (highest to lowest): 

53 1. Runtime overrides (hot-reloadable) 

54 2. Environment variables (LEX_ADMIN__*) 

55 3. YAML file configuration 

56 4. Pydantic model defaults 

57 

58 Usage: 

59 loader = AdminConfigLoader() 

60 config = await loader.load() 

61 

62 Hot reload: 

63 await loader.reload() 

64 

65 Environment variable mapping: 

66 admin.auth.session_lifetime -> LEX_ADMIN__AUTH__SESSION_LIFETIME 

67 """ 

68 

69 ENV_PREFIX = "LEX_ADMIN__" 

70 YAML_SECTION = "admin" 

71 

72 def __init__( 

73 self, 

74 yaml_path: Path | None = None, 

75 base_loader: BaseConfigLoader | None = None, 

76 defaults: dict[str, Any] | None = None, 

77 ): 

78 self._yaml_path = yaml_path 

79 self._base_loader = base_loader 

80 self._defaults = defaults or {} 

81 self._config: AdminConfig | None = None 

82 self._runtime_overrides: dict[str, Any] = {} 

83 self._reload_callbacks: list[Callable[[AdminConfig], None]] = [] 

84 

85 @property 

86 def config(self) -> AdminConfig: 

87 """Get current config (raises if not loaded).""" 

88 if self._config is None: 

89 raise RuntimeError( 

90 "Config not loaded. Call load() first or use get_config().", 

91 ) 

92 return self._config 

93 

94 async def load(self) -> AdminConfig: 

95 """Load configuration from all sources. 

96 

97 Returns: 

98 Merged AdminConfig instance 

99 """ 

100 # Layer 1: Start with defaults from Pydantic model 

101 config_data: dict[str, Any] = {} 

102 

103 # Layer 2: Load from YAML 

104 yaml_data = await self._load_yaml() 

105 self._deep_merge(config_data, yaml_data) 

106 

107 # Layer 3: Apply environment variables 

108 env_data = self._load_env() 

109 self._deep_merge(config_data, env_data) 

110 

111 # Layer 4: Apply constructor defaults 

112 self._deep_merge(config_data, self._defaults) 

113 

114 # Layer 5: Apply runtime overrides 

115 self._deep_merge(config_data, self._runtime_overrides) 

116 

117 # Construct config model 

118 self._config = AdminConfig.model_validate(config_data) 

119 return self._config 

120 

121 async def reload(self) -> AdminConfig: 

122 """Hot-reload configuration. 

123 

124 Re-reads YAML and environment, preserving runtime overrides. 

125 Notifies all registered reload callbacks. 

126 

127 Returns: 

128 Newly loaded AdminConfig instance 

129 """ 

130 old_config = self._config 

131 new_config = await self.load() 

132 

133 # Notify callbacks if config changed 

134 if old_config != new_config: 

135 for callback in self._reload_callbacks: 

136 callback(new_config) 

137 

138 return new_config 

139 

140 def on_reload(self, callback: Callable[[AdminConfig], None]) -> None: 

141 """Register a callback to be notified on config reload.""" 

142 self._reload_callbacks.append(callback) 

143 

144 def set_runtime(self, path: str, value: Any) -> None: 

145 """Set a runtime override (survives reload). 

146 

147 Args: 

148 path: Dot-separated config path (e.g., "auth.session_lifetime") 

149 value: Value to set 

150 """ 

151 self._set_nested(self._runtime_overrides, path.split("."), value) 

152 

153 def clear_runtime(self, path: str | None = None) -> None: 

154 """Clear runtime overrides. 

155 

156 Args: 

157 path: Specific path to clear, or None for all 

158 """ 

159 if path is None: 

160 self._runtime_overrides.clear() 

161 else: 

162 self._delete_nested(self._runtime_overrides, path.split(".")) 

163 

164 def get(self, path: str, default: T = None) -> T | Any: # type: ignore[assignment] 

165 """Get a config value by dot-path. 

166 

167 Args: 

168 path: Dot-separated path (e.g., "auth.session_lifetime") 

169 default: Default if path not found 

170 

171 Returns: 

172 Config value or default 

173 """ 

174 if self._config is None: 

175 return default 

176 

177 obj: Any = self._config 

178 for part in path.split("."): 

179 if hasattr(obj, part): 

180 obj = getattr(obj, part) 

181 elif isinstance(obj, dict) and part in obj: 

182 obj = obj[part] 

183 else: 

184 return default 

185 return obj 

186 

187 async def _load_yaml(self) -> dict[str, Any]: 

188 """Load configuration from YAML file. 

189 

190 Tries paths in order: 

191 1. Explicit yaml_path if provided 

192 2. Common paths: application.yaml, application.yml, config/application.yaml 

193 

194 Returns: 

195 Dict of admin configuration from YAML, or empty dict 

196 """ 

197 paths_to_try = [] 

198 

199 if self._yaml_path: 

200 paths_to_try.append(self._yaml_path) 

201 

202 # Common paths 

203 paths_to_try.extend( 

204 [ 

205 Path.cwd() / "application.yaml", 

206 Path.cwd() / "application.yml", 

207 Path.cwd() / "config" / "application.yaml", 

208 ], 

209 ) 

210 

211 for path in paths_to_try: 

212 if path.exists(): 

213 try: 

214 async with aiofiles.open(path) as f: 

215 content = await f.read() 

216 data = yaml.safe_load(content) or {} 

217 return data.get(self.YAML_SECTION, {}) 

218 except (yaml.YAMLError, ValueError, OSError): 

219 continue 

220 

221 return {} 

222 

223 def _load_env(self) -> dict[str, Any]: 

224 """Load configuration from environment variables. 

225 

226 Maps LEX_ADMIN__* environment variables to nested config: 

227 - LEX_ADMIN__DEBUG=true -> {"debug": True} 

228 - LEX_ADMIN__AUTH__SESSION_LIFETIME=3600 -> {"auth": {"session_lifetime": 3600}} 

229 """ 

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

231 

232 for key, value in os.environ.items(): 

233 if not key.startswith(self.ENV_PREFIX): 

234 continue 

235 

236 # Remove prefix and convert to path 

237 config_key = key[len(self.ENV_PREFIX) :] 

238 if "__" in config_key: 

239 # Canonical format: LEX_ADMIN__AUTH__SESSION_LIFETIME 

240 resolved_path = [ 

241 segment.lower() for segment in config_key.split("__") if segment 

242 ] 

243 else: 

244 # Backward-compat fallback: LEX_ADMIN_AUTH_SESSION_LIFETIME 

245 path_parts = config_key.lower().split("_") 

246 resolved_path = self._resolve_env_path(path_parts) 

247 

248 # Convert value 

249 typed_value = self._parse_env_value(value) 

250 

251 self._set_nested(result, resolved_path, typed_value) 

252 

253 return result 

254 

255 def _resolve_env_path(self, parts: list[str]) -> list[str]: 

256 """Resolve environment variable parts to config path. 

257 

258 Handles compound names like SESSION_LIFETIME -> session_lifetime 

259 """ 

260 # Known nested sections 

261 known_sections = {"auth", "features", "ui", "rate", "resource", "table", "form"} 

262 

263 result = [] 

264 i = 0 

265 while i < len(parts): 

266 part = parts[i] 

267 

268 if part in known_sections: 

269 # Check for compound section name 

270 if part == "rate" and i + 1 < len(parts) and parts[i + 1] == "limit": 

271 result.append("rate_limit") 

272 i += 2 

273 elif ( 

274 part == "resource" 

275 and i + 1 < len(parts) 

276 and parts[i + 1] == "defaults" 

277 ): 

278 result.append("resource_defaults") 

279 i += 2 

280 elif ( 

281 part == "table" 

282 and i + 1 < len(parts) 

283 and parts[i + 1] == "defaults" 

284 ): 

285 result.append("table_defaults") 

286 i += 2 

287 elif ( 

288 part == "form" and i + 1 < len(parts) and parts[i + 1] == "defaults" 

289 ): 

290 result.append("form_defaults") 

291 i += 2 

292 else: 

293 result.append(part) 

294 i += 1 

295 else: 

296 # Remaining parts as compound key 

297 result.append("_".join(parts[i:])) 

298 break 

299 

300 return result if result else parts 

301 

302 def _parse_env_value(self, value: str) -> Any: 

303 """Parse environment variable string to typed value.""" 

304 # Boolean 

305 if value.lower() in ("true", "1", "yes", "on"): 

306 return True 

307 if value.lower() in ("false", "0", "no", "off"): 

308 return False 

309 

310 # Integer 

311 try: 

312 return int(value) 

313 except ValueError: 

314 pass 

315 

316 # Float 

317 try: 

318 return float(value) 

319 except ValueError: 

320 pass 

321 

322 # JSON (for complex values) 

323 if value.startswith(("{", "[")): 

324 try: 

325 from lexigram.serialization import loads 

326 

327 return loads(value) 

328 except (json.JSONDecodeError, ValueError, TypeError): 

329 pass 

330 

331 # String 

332 return value 

333 

334 def _deep_merge(self, base: dict[str, Any], override: dict[str, Any]) -> None: 

335 """Deep merge override into base (mutates base).""" 

336 for key, value in override.items(): 

337 if key in base and isinstance(base[key], dict) and isinstance(value, dict): 

338 self._deep_merge(base[key], value) 

339 else: 

340 base[key] = value 

341 

342 def _set_nested(self, data: dict[str, Any], path: list[str], value: Any) -> None: 

343 """Set a nested value by path.""" 

344 for part in path[:-1]: 

345 data = data.setdefault(part, {}) 

346 data[path[-1]] = value 

347 

348 def _delete_nested(self, data: dict[str, Any], path: list[str]) -> None: 

349 """Delete a nested value by path.""" 

350 for part in path[:-1]: 

351 if part in data and isinstance(data[part], dict): 

352 data = data[part] 

353 else: 

354 return 

355 data.pop(path[-1], None) 

356 

357 

358# Global config instance (lazily initialized) 

359_config_loader: AdminConfigLoader | None = None 

360 

361 

362async def get_config() -> AdminConfig: 

363 """Get the global admin configuration. 

364 

365 Initializes and loads config on first call. 

366 

367 Returns: 

368 AdminConfig instance 

369 """ 

370 global _config_loader 

371 if _config_loader is None: 

372 _config_loader = AdminConfigLoader() 

373 await _config_loader.load() 

374 return _config_loader.config 

375 

376 

377def get_loader() -> AdminConfigLoader: 

378 """Get the global config loader instance. 

379 

380 Creates a new loader if none exists. 

381 

382 Returns: 

383 AdminConfigLoader instance 

384 """ 

385 global _config_loader 

386 if _config_loader is None: 

387 _config_loader = AdminConfigLoader() 

388 return _config_loader