Coverage for src / ezplog / config / manager.py: 92.26%

1# /////////////////////////////////////////////////////////////// 

2# EZPL - Configuration Manager 

3# Project: ezpl 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Configuration manager for Ezpl logging framework. 

8 

9This module provides centralized configuration management with support for 

10file-based configuration, environment variables, and runtime configuration. 

11""" 

12 

13from __future__ import annotations 

14 

15# /////////////////////////////////////////////////////////////// 

16# IMPORTS 

17# /////////////////////////////////////////////////////////////// 

18# Standard library imports 

19import json 

20import os 

21import shlex 

22import warnings 

23from pathlib import Path 

24from typing import Any, cast 

25 

26# Local imports 

27from ..core.exceptions import FileOperationError, ValidationError 

28from ._defaults import DefaultConfiguration 

29 

30# /////////////////////////////////////////////////////////////// 

31# CLASSES 

32# /////////////////////////////////////////////////////////////// 

33 

34 

35class ConfigurationManager: 

36 """ 

37 Centralized configuration manager for Ezpl. 

38 

39 This class handles all configuration operations including loading, 

40 saving, and merging configuration from multiple sources. 

41 """ 

42 

43 _ENV_MAPPINGS = { 

44 "EZPL_LOG_LEVEL": "log-level", 

45 "EZPL_LOG_FILE": "log-file", 

46 "EZPL_LOG_DIR": "log-dir", 

47 "EZPL_PRINTER_LEVEL": "printer-level", 

48 "EZPL_INDENT_STEP": "indent-step", 

49 "EZPL_INDENT_SYMBOL": "indent-symbol", 

50 "EZPL_BASE_INDENT_SYMBOL": "base-indent-symbol", 

51 "EZPL_FILE_LOGGER_LEVEL": "file-logger-level", 

52 "EZPL_LOG_FORMAT": "log-format", 

53 "EZPL_LOG_ROTATION": "log-rotation", 

54 "EZPL_LOG_RETENTION": "log-retention", 

55 "EZPL_LOG_COMPRESSION": "log-compression", 

56 } 

57 

58 # /////////////////////////////////////////////////////////////// 

59 # INIT 

60 # /////////////////////////////////////////////////////////////// 

61 

62 def __init__(self, config_file: Path | None = None): 

63 """ 

64 Initialize the configuration manager. 

65 

66 Args: 

67 config_file: Optional path to configuration file. 

68 Defaults to ~/.ezpl/config.json 

69 """ 

70 self._config_file = config_file or DefaultConfiguration.CONFIG_FILE 

71 self._config: dict[str, Any] = {} 

72 self._explicit_keys: set[str] = set() 

73 self._load_configuration() 

74 

75 # ------------------------------------------------ 

76 # PRIVATE HELPER METHODS 

77 # ------------------------------------------------ 

78 

79 def _load_configuration(self) -> None: 

80 """ 

81 Load configuration from file and environment variables. 

82 

83 Priority order: 

84 1. Environment variables (highest priority) 

85 2. Configuration file 

86 3. Default values (lowest priority) 

87 """ 

88 # Start with defaults — these are never considered "explicit" 

89 self._config = DefaultConfiguration.get_all_defaults().copy() 

90 self._explicit_keys = set() 

91 

92 # Load from file if it exists — keys from file are explicit 

93 if self._config_file.exists(): 

94 try: 

95 with open(self._config_file, encoding="utf-8") as f: 

96 file_config = json.load(f) 

97 self._config.update(file_config) 

98 self._explicit_keys.update(file_config.keys()) 

99 except (OSError, json.JSONDecodeError) as e: 

100 # If file is corrupted, use defaults 

101 warnings.warn( 

102 f"Could not load config file {self._config_file}: {e}", 

103 UserWarning, 

104 stacklevel=2, 

105 ) 

106 

107 # Override with environment variables — env keys are explicit 

108 self._load_from_environment() 

109 

110 def _load_from_environment(self) -> None: 

111 """ 

112 Load configuration from environment variables. 

113 

114 Environment variables should be prefixed with 'EZPL_' and use 

115 uppercase with underscores (e.g., EZPL_LOG_LEVEL). 

116 """ 

117 user_env_vars = self._load_user_env_file() 

118 

119 for env_var, config_key in self._ENV_MAPPINGS.items(): 

120 value = os.getenv(env_var) 

121 if value is None: 

122 value = user_env_vars.get(env_var) 

123 if value is not None: 

124 # Convert string values to appropriate types 

125 if config_key in ["indent-step"]: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true

126 try: 

127 self._config[config_key] = int(value) 

128 except ValueError as e: 

129 raise ValidationError( 

130 f"Environment variable {env_var} must be an integer, got: {value!r}", 

131 env_var, 

132 value, 

133 ) from e 

134 else: 

135 self._config[config_key] = value 

136 self._explicit_keys.add(config_key) 

137 

138 def _load_user_env_file(self) -> dict[str, str]: 

139 """ 

140 Load user-level EZPL environment variables from ~/.ezpl/.env. 

141 

142 Returns: 

143 Dictionary of environment variables found in user env file. 

144 """ 

145 env_file = DefaultConfiguration.CONFIG_DIR / ".env" 

146 env_vars: dict[str, str] = {} 

147 

148 if not env_file.exists(): 

149 return env_vars 

150 

151 try: 

152 with open(env_file, encoding="utf-8") as f: 

153 for line in f: 

154 stripped = line.strip() 

155 if not stripped or stripped.startswith("#") or "=" not in stripped: 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true

156 continue 

157 key, value = stripped.split("=", 1) 

158 env_vars[key.strip()] = value.strip() 

159 except OSError: 

160 return {} 

161 

162 return env_vars 

163 

164 # /////////////////////////////////////////////////////////////// 

165 # GETTER 

166 # /////////////////////////////////////////////////////////////// 

167 

168 @property 

169 def config_file(self) -> Path: 

170 """Return the path to the configuration file.""" 

171 return self._config_file 

172 

173 def get(self, key: str, default: Any = None) -> Any: 

174 """ 

175 Get a configuration value. 

176 

177 Args: 

178 key: Configuration key 

179 default: Default value if key not found 

180 

181 Returns: 

182 Configuration value or default 

183 """ 

184 return self._config.get(key, default) 

185 

186 def has_key(self, key: str) -> bool: 

187 """ 

188 Check if a configuration key is explicitly set (not just a default). 

189 

190 A key is considered explicit when it comes from a config file, an 

191 environment variable, or a direct call to configure()/set()/update(). 

192 Keys that are present only because of default values return False. 

193 

194 Args: 

195 key: Configuration key to check 

196 

197 Returns: 

198 True if the key was explicitly set, False if it is only a default. 

199 """ 

200 return key in self._explicit_keys 

201 

202 def get_log_level(self) -> str: 

203 """Get the current log level.""" 

204 return cast(str, self.get("log-level", DefaultConfiguration.LOG_LEVEL)) 

205 

206 def get_log_file(self) -> Path: 

207 """Get the current log file path.""" 

208 log_file = self.get("log-file", DefaultConfiguration.LOG_FILE) 

209 log_dir = self.get("log-dir", DefaultConfiguration.LOG_DIR) 

210 

211 # Convert to Path if string 

212 log_file_path = Path(log_file) if isinstance(log_file, str) else log_file 

213 log_dir_path = Path(log_dir) if isinstance(log_dir, str) else log_dir 

214 

215 # If log_file is relative, make it relative to log_dir 

216 if not log_file_path.is_absolute(): 216 ↛ 218line 216 didn't jump to line 218 because the condition on line 216 was always true

217 return log_dir_path / log_file_path 

218 return log_file_path 

219 

220 def get_printer_level(self) -> str: 

221 """Get the current printer level.""" 

222 return cast(str, self.get("printer-level", DefaultConfiguration.PRINTER_LEVEL)) 

223 

224 def get_file_logger_level(self) -> str: 

225 """Get the current file logger level.""" 

226 return cast( 

227 str, self.get("file-logger-level", DefaultConfiguration.FILE_LOGGER_LEVEL) 

228 ) 

229 

230 def get_indent_step(self) -> int: 

231 """Get the current indent step.""" 

232 return cast(int, self.get("indent-step", DefaultConfiguration.INDENT_STEP)) 

233 

234 def get_indent_symbol(self) -> str: 

235 """Get the current indent symbol.""" 

236 return cast(str, self.get("indent-symbol", DefaultConfiguration.INDENT_SYMBOL)) 

237 

238 def get_base_indent_symbol(self) -> str: 

239 """Get the current base indent symbol.""" 

240 return cast( 

241 str, self.get("base-indent-symbol", DefaultConfiguration.BASE_INDENT_SYMBOL) 

242 ) 

243 

244 def get_log_format(self) -> str: 

245 """Get the current log format.""" 

246 return cast(str, self.get("log-format", DefaultConfiguration._LOG_FORMAT)) 

247 

248 def get_log_rotation(self) -> str | None: 

249 """Get the current log rotation setting.""" 

250 return cast( 

251 str | None, self.get("log-rotation", DefaultConfiguration.LOG_ROTATION) 

252 ) 

253 

254 def get_log_retention(self) -> str | None: 

255 """Get the current log retention setting.""" 

256 return cast( 

257 str | None, self.get("log-retention", DefaultConfiguration.LOG_RETENTION) 

258 ) 

259 

260 def get_log_compression(self) -> str | None: 

261 """Get the current log compression setting.""" 

262 return cast( 

263 str | None, 

264 self.get("log-compression", DefaultConfiguration.LOG_COMPRESSION), 

265 ) 

266 

267 def get_all(self) -> dict[str, Any]: 

268 """ 

269 Get all configuration values. 

270 

271 Returns: 

272 Dictionary containing all configuration values 

273 """ 

274 return self._config.copy() 

275 

276 # /////////////////////////////////////////////////////////////// 

277 # SETTER 

278 # /////////////////////////////////////////////////////////////// 

279 

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

281 """ 

282 Set a configuration value. 

283 

284 Args: 

285 key: Configuration key 

286 value: Configuration value 

287 """ 

288 self._config[key] = value 

289 self._explicit_keys.add(key) 

290 

291 def update(self, config_dict: dict[str, Any]) -> None: 

292 """ 

293 Update configuration with new values. 

294 

295 Args: 

296 config_dict: Dictionary of configuration values to update 

297 """ 

298 self._config.update(config_dict) 

299 self._explicit_keys.update(config_dict.keys()) 

300 

301 # /////////////////////////////////////////////////////////////// 

302 # FILE OPERATIONS 

303 # /////////////////////////////////////////////////////////////// 

304 

305 def save(self) -> None: 

306 """ 

307 Save current configuration to file. 

308 

309 Raises: 

310 FileOperationError: If unable to write to configuration file 

311 """ 

312 try: 

313 # Ensure config directory exists 

314 self._config_file.parent.mkdir(parents=True, exist_ok=True) 

315 

316 # Save configuration 

317 with open(self._config_file, "w", encoding="utf-8") as f: 

318 json.dump(self._config, f, indent=4, ensure_ascii=False) 

319 except OSError as e: 

320 raise FileOperationError( 

321 f"Could not save configuration to {self._config_file}: {e}", 

322 str(self._config_file), 

323 "save", 

324 ) from e 

325 

326 def reset_to_defaults(self) -> None: 

327 """Reset configuration to default values.""" 

328 self._config = DefaultConfiguration.get_all_defaults().copy() 

329 self._explicit_keys = set() 

330 

331 def reload(self) -> None: 

332 """ 

333 Reload configuration from file and environment variables. 

334 

335 This method reloads the configuration, useful when environment 

336 variables or the config file have changed after initialization. 

337 """ 

338 self._load_configuration() 

339 

340 def export_to_script( 

341 self, output_file: str | Path, platform: str | None = None 

342 ) -> None: 

343 """ 

344 Export configuration as environment variables script. 

345 

346 Args: 

347 output_file: Path to output script file 

348 platform: Target platform ('windows', 'unix', or None for auto-detect) 

349 

350 Raises: 

351 FileOperationError: If unable to write to output file 

352 """ 

353 if platform is None: 

354 import sys 

355 

356 platform = "windows" if sys.platform == "win32" else "unix" 

357 

358 output_path = Path(output_file) 

359 

360 try: 

361 with open(output_path, "w", encoding="utf-8") as f: 

362 if platform == "windows": 

363 # Generate Batch script for Windows 

364 for env_var, config_key in self._ENV_MAPPINGS.items(): 

365 value = self._config.get(config_key) 

366 if value is None: 

367 continue 

368 safe_value = str(value).replace('"', '""') 

369 f.write(f'set "{env_var}={safe_value}"\n') 

370 else: 

371 # Generate Bash script for Unix/Linux/macOS 

372 f.write("#!/bin/bash\n") 

373 for env_var, config_key in self._ENV_MAPPINGS.items(): 

374 value = self._config.get(config_key) 

375 if value is None: 

376 continue 

377 f.write(f"export {env_var}={shlex.quote(str(value))}\n") 

378 except OSError as e: 

379 raise FileOperationError( 

380 f"Could not write to {output_path}: {e}", 

381 str(output_path), 

382 "export", 

383 ) from e 

384 

385 # /////////////////////////////////////////////////////////////// 

386 # REPRESENTATION METHODS 

387 # /////////////////////////////////////////////////////////////// 

388 

389 def __str__(self) -> str: 

390 """String representation of the configuration.""" 

391 return f"ConfigurationManager(config_file={self._config_file})" 

392 

393 def __repr__(self) -> str: 

394 """Detailed string representation of the configuration.""" 

395 return f"ConfigurationManager(config_file={self._config_file}, config={self._config})" 

396 

397 

398# /////////////////////////////////////////////////////////////// 

399# PUBLIC API 

400# /////////////////////////////////////////////////////////////// 

401 

402__all__ = [ 

403 "ConfigurationManager", 

404]