Coverage for src/ezplog/config/manager.py: 92.44%
142 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-16 19:04 +0000
1# ///////////////////////////////////////////////////////////////
2# EZPL - Configuration Manager
3# Project: ezpl
4# ///////////////////////////////////////////////////////////////
6"""
7Configuration manager for Ezpl logging framework.
9This module provides centralized configuration management with support for
10file-based configuration, environment variables, and runtime configuration.
11"""
13from __future__ import annotations
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
26# Local imports
27from ..core.exceptions import FileOperationError, ValidationError
28from ._defaults import DefaultConfiguration
30# ///////////////////////////////////////////////////////////////
31# CLASSES
32# ///////////////////////////////////////////////////////////////
35class ConfigurationManager:
36 """
37 Centralized configuration manager for Ezpl.
39 This class handles all configuration operations including loading,
40 saving, and merging configuration from multiple sources.
41 """
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 }
58 # ///////////////////////////////////////////////////////////////
59 # INIT
60 # ///////////////////////////////////////////////////////////////
62 def __init__(self, config_file: Path | None = None):
63 """
64 Initialize the configuration manager.
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()
75 # ------------------------------------------------
76 # PRIVATE HELPER METHODS
77 # ------------------------------------------------
79 def _load_configuration(self) -> None:
80 """
81 Load configuration from file and environment variables.
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()
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 )
107 # Override with environment variables — env keys are explicit
108 self._load_from_environment()
110 def _load_from_environment(self) -> None:
111 """
112 Load configuration from environment variables.
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()
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)
138 def _load_user_env_file(self) -> dict[str, str]:
139 """
140 Load user-level EZPL environment variables from ~/.ezpl/.env.
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] = {}
148 if not env_file.exists():
149 return env_vars
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 {}
162 return env_vars
164 # ///////////////////////////////////////////////////////////////
165 # GETTER
166 # ///////////////////////////////////////////////////////////////
168 @property
169 def config_file(self) -> Path:
170 """Return the path to the configuration file."""
171 return self._config_file
173 def get(self, key: str, default: Any = None) -> Any:
174 """
175 Get a configuration value.
177 Args:
178 key: Configuration key
179 default: Default value if key not found
181 Returns:
182 Configuration value or default
183 """
184 return self._config.get(key, default)
186 def has_key(self, key: str) -> bool:
187 """
188 Check if a configuration key is explicitly set (not just a default).
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.
194 Args:
195 key: Configuration key to check
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
202 def get_log_level(self) -> str:
203 """Get the current log level."""
204 return cast(str, self.get("log-level", DefaultConfiguration.LOG_LEVEL))
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)
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
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
220 def get_printer_level(self) -> str:
221 """Get the current printer level."""
222 return cast(str, self.get("printer-level", DefaultConfiguration.PRINTER_LEVEL))
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 )
230 def get_indent_step(self) -> int:
231 """Get the current indent step."""
232 return cast(int, self.get("indent-step", DefaultConfiguration.INDENT_STEP))
234 def get_indent_symbol(self) -> str:
235 """Get the current indent symbol."""
236 return cast(str, self.get("indent-symbol", DefaultConfiguration.INDENT_SYMBOL))
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 )
244 def get_log_format(self) -> str:
245 """Get the current log format."""
246 return cast(str, self.get("log-format", DefaultConfiguration._LOG_FORMAT))
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 )
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 )
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 )
267 def get_log_backtrace(self) -> bool:
268 """Get the current log backtrace setting."""
269 return cast(bool, self.get("log-backtrace", DefaultConfiguration.LOG_BACKTRACE))
271 def get_log_diagnose(self) -> bool:
272 """Get the current log diagnose setting."""
273 return cast(bool, self.get("log-diagnose", DefaultConfiguration.LOG_DIAGNOSE))
275 def get_all(self) -> dict[str, Any]:
276 """
277 Get all configuration values.
279 Returns:
280 Dictionary containing all configuration values
281 """
282 return self._config.copy()
284 # ///////////////////////////////////////////////////////////////
285 # SETTER
286 # ///////////////////////////////////////////////////////////////
288 def set(self, key: str, value: Any) -> None:
289 """
290 Set a configuration value.
292 Args:
293 key: Configuration key
294 value: Configuration value
295 """
296 self._config[key] = value
297 self._explicit_keys.add(key)
299 def update(self, config_dict: dict[str, Any]) -> None:
300 """
301 Update configuration with new values.
303 Args:
304 config_dict: Dictionary of configuration values to update
305 """
306 self._config.update(config_dict)
307 self._explicit_keys.update(config_dict.keys())
309 # ///////////////////////////////////////////////////////////////
310 # FILE OPERATIONS
311 # ///////////////////////////////////////////////////////////////
313 def save(self) -> None:
314 """
315 Save current configuration to file.
317 Raises:
318 FileOperationError: If unable to write to configuration file
319 """
320 try:
321 # Ensure config directory exists
322 self._config_file.parent.mkdir(parents=True, exist_ok=True)
324 # Save configuration
325 with open(self._config_file, "w", encoding="utf-8") as f:
326 json.dump(self._config, f, indent=4, ensure_ascii=False)
327 except OSError as e:
328 raise FileOperationError(
329 f"Could not save configuration to {self._config_file}: {e}",
330 str(self._config_file),
331 "save",
332 ) from e
334 def reset_to_defaults(self) -> None:
335 """Reset configuration to default values."""
336 self._config = DefaultConfiguration.get_all_defaults().copy()
337 self._explicit_keys = set()
339 def reload(self) -> None:
340 """
341 Reload configuration from file and environment variables.
343 This method reloads the configuration, useful when environment
344 variables or the config file have changed after initialization.
345 """
346 self._load_configuration()
348 def export_to_script(
349 self, output_file: str | Path, platform: str | None = None
350 ) -> None:
351 """
352 Export configuration as environment variables script.
354 Args:
355 output_file: Path to output script file
356 platform: Target platform ('windows', 'unix', or None for auto-detect)
358 Raises:
359 FileOperationError: If unable to write to output file
360 """
361 if platform is None:
362 import sys
364 platform = "windows" if sys.platform == "win32" else "unix"
366 output_path = Path(output_file)
368 try:
369 with open(output_path, "w", encoding="utf-8") as f:
370 if platform == "windows":
371 # Generate Batch script for Windows
372 for env_var, config_key in self._ENV_MAPPINGS.items():
373 value = self._config.get(config_key)
374 if value is None:
375 continue
376 safe_value = str(value).replace('"', '""')
377 f.write(f'set "{env_var}={safe_value}"\n')
378 else:
379 # Generate Bash script for Unix/Linux/macOS
380 f.write("#!/bin/bash\n")
381 for env_var, config_key in self._ENV_MAPPINGS.items():
382 value = self._config.get(config_key)
383 if value is None:
384 continue
385 f.write(f"export {env_var}={shlex.quote(str(value))}\n")
386 except OSError as e:
387 raise FileOperationError(
388 f"Could not write to {output_path}: {e}",
389 str(output_path),
390 "export",
391 ) from e
393 # ///////////////////////////////////////////////////////////////
394 # REPRESENTATION METHODS
395 # ///////////////////////////////////////////////////////////////
397 def __str__(self) -> str:
398 """String representation of the configuration."""
399 return f"ConfigurationManager(config_file={self._config_file})"
401 def __repr__(self) -> str:
402 """Detailed string representation of the configuration."""
403 return f"ConfigurationManager(config_file={self._config_file}, config={self._config})"
406# ///////////////////////////////////////////////////////////////
407# PUBLIC API
408# ///////////////////////////////////////////////////////////////
410__all__ = [
411 "ConfigurationManager",
412]