Coverage for src / ezpl / config / manager.py: 90.24%
132 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-13 19:35 +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
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._load_configuration()
74 # ------------------------------------------------
75 # PRIVATE HELPER METHODS
76 # ------------------------------------------------
78 def _load_configuration(self) -> None:
79 """
80 Load configuration from file and environment variables.
82 Priority order:
83 1. Environment variables (highest priority)
84 2. Configuration file
85 3. Default values (lowest priority)
86 """
87 # Start with defaults
88 self._config = DefaultConfiguration.get_all_defaults().copy()
90 # Load from file if it exists
91 if self._config_file.exists():
92 try:
93 with open(self._config_file, encoding="utf-8") as f:
94 file_config = json.load(f)
95 self._config.update(file_config)
96 except (OSError, json.JSONDecodeError) as e:
97 # If file is corrupted, use defaults
98 warnings.warn(
99 f"Could not load config file {self._config_file}: {e}",
100 UserWarning,
101 stacklevel=2,
102 )
104 # Override with environment variables
105 self._load_from_environment()
107 def _load_from_environment(self) -> None:
108 """
109 Load configuration from environment variables.
111 Environment variables should be prefixed with 'EZPL_' and use
112 uppercase with underscores (e.g., EZPL_LOG_LEVEL).
113 """
114 user_env_vars = self._load_user_env_file()
116 for env_var, config_key in self.ENV_MAPPINGS.items():
117 value = os.getenv(env_var)
118 if value is None:
119 value = user_env_vars.get(env_var)
120 if value is not None:
121 # Convert string values to appropriate types
122 if config_key in ["indent-step"]: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 try:
124 self._config[config_key] = int(value)
125 except ValueError as e:
126 raise ValueError(
127 f"Failed to convert {value} to int: {e}"
128 ) from e
129 else:
130 self._config[config_key] = value
132 def _load_user_env_file(self) -> dict[str, str]:
133 """
134 Load user-level EZPL environment variables from ~/.ezpl/.env.
136 Returns:
137 Dictionary of environment variables found in user env file.
138 """
139 env_file = DefaultConfiguration.CONFIG_DIR / ".env"
140 env_vars: dict[str, str] = {}
142 if not env_file.exists():
143 return env_vars
145 try:
146 with open(env_file, encoding="utf-8") as f:
147 for line in f:
148 stripped = line.strip()
149 if not stripped or stripped.startswith("#") or "=" not in stripped: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 continue
151 key, value = stripped.split("=", 1)
152 env_vars[key.strip()] = value.strip()
153 except OSError:
154 return {}
156 return env_vars
158 # ///////////////////////////////////////////////////////////////
159 # GETTER
160 # ///////////////////////////////////////////////////////////////
162 @property
163 def config_file(self) -> Path:
164 """Return the path to the configuration file."""
165 return self._config_file
167 def get(self, key: str, default: Any = None) -> Any:
168 """
169 Get a configuration value.
171 Args:
172 key: Configuration key
173 default: Default value if key not found
175 Returns:
176 Configuration value or default
177 """
178 return self._config.get(key, default)
180 def has_key(self, key: str) -> bool:
181 """
182 Check if a configuration key is explicitly set (not just a default).
184 Args:
185 key: Configuration key to check
187 Returns:
188 True if the key is explicitly set in config or environment, False otherwise
189 """
190 # Check if key exists in config (from file or explicitly set)
191 if key in self._config: 191 ↛ 194line 191 didn't jump to line 194 because the condition on line 191 was always true
192 return True
193 # Check if corresponding environment variable exists
194 env_key = f"EZPL_{key.replace('-', '_').upper()}"
195 return env_key in os.environ
197 def get_log_level(self) -> str:
198 """Get the current log level."""
199 return cast(str, self.get("log-level", DefaultConfiguration.LOG_LEVEL))
201 def get_log_file(self) -> Path:
202 """Get the current log file path."""
203 log_file = self.get("log-file", DefaultConfiguration.LOG_FILE)
204 log_dir = self.get("log-dir", DefaultConfiguration.LOG_DIR)
206 # Convert to Path if string
207 log_file_path = Path(log_file) if isinstance(log_file, str) else log_file
208 log_dir_path = Path(log_dir) if isinstance(log_dir, str) else log_dir
210 # If log_file is relative, make it relative to log_dir
211 if not log_file_path.is_absolute(): 211 ↛ 213line 211 didn't jump to line 213 because the condition on line 211 was always true
212 return log_dir_path / log_file_path
213 return log_file_path
215 def get_printer_level(self) -> str:
216 """Get the current printer level."""
217 return cast(str, self.get("printer-level", DefaultConfiguration.PRINTER_LEVEL))
219 def get_file_logger_level(self) -> str:
220 """Get the current file logger level."""
221 return cast(
222 str, self.get("file-logger-level", DefaultConfiguration.FILE_LOGGER_LEVEL)
223 )
225 def get_indent_step(self) -> int:
226 """Get the current indent step."""
227 return cast(int, self.get("indent-step", DefaultConfiguration.INDENT_STEP))
229 def get_indent_symbol(self) -> str:
230 """Get the current indent symbol."""
231 return cast(str, self.get("indent-symbol", DefaultConfiguration.INDENT_SYMBOL))
233 def get_base_indent_symbol(self) -> str:
234 """Get the current base indent symbol."""
235 return cast(
236 str, self.get("base-indent-symbol", DefaultConfiguration.BASE_INDENT_SYMBOL)
237 )
239 def get_log_format(self) -> str:
240 """Get the current log format."""
241 return cast(str, self.get("log-format", DefaultConfiguration.LOG_FORMAT))
243 def get_log_rotation(self) -> str | None:
244 """Get the current log rotation setting."""
245 return cast(
246 str | None, self.get("log-rotation", DefaultConfiguration.LOG_ROTATION)
247 )
249 def get_log_retention(self) -> str | None:
250 """Get the current log retention setting."""
251 return cast(
252 str | None, self.get("log-retention", DefaultConfiguration.LOG_RETENTION)
253 )
255 def get_log_compression(self) -> str | None:
256 """Get the current log compression setting."""
257 return cast(
258 str | None,
259 self.get("log-compression", DefaultConfiguration.LOG_COMPRESSION),
260 )
262 def get_all(self) -> dict[str, Any]:
263 """
264 Get all configuration values.
266 Returns:
267 Dictionary containing all configuration values
268 """
269 return self._config.copy()
271 # ///////////////////////////////////////////////////////////////
272 # SETTER
273 # ///////////////////////////////////////////////////////////////
275 def set(self, key: str, value: Any) -> None:
276 """
277 Set a configuration value.
279 Args:
280 key: Configuration key
281 value: Configuration value
282 """
283 self._config[key] = value
285 def update(self, config_dict: dict[str, Any]) -> None:
286 """
287 Update configuration with new values.
289 Args:
290 config_dict: Dictionary of configuration values to update
291 """
292 self._config.update(config_dict)
294 # ///////////////////////////////////////////////////////////////
295 # FILE OPERATIONS
296 # ///////////////////////////////////////////////////////////////
298 def save(self) -> None:
299 """
300 Save current configuration to file.
302 Raises:
303 FileOperationError: If unable to write to configuration file
304 """
305 try:
306 # Ensure config directory exists
307 self._config_file.parent.mkdir(parents=True, exist_ok=True)
309 # Save configuration
310 with open(self._config_file, "w", encoding="utf-8") as f:
311 json.dump(self._config, f, indent=4, ensure_ascii=False)
312 except OSError as e:
313 raise FileOperationError(
314 f"Could not save configuration to {self._config_file}: {e}",
315 str(self._config_file),
316 "save",
317 ) from e
319 def reset_to_defaults(self) -> None:
320 """Reset configuration to default values."""
321 self._config = DefaultConfiguration.get_all_defaults().copy()
323 def reload(self) -> None:
324 """
325 Reload configuration from file and environment variables.
327 This method reloads the configuration, useful when environment
328 variables or the config file have changed after initialization.
329 """
330 self._load_configuration()
332 def export_to_script(
333 self, output_file: str | Path, platform: str | None = None
334 ) -> None:
335 """
336 Export configuration as environment variables script.
338 Args:
339 output_file: Path to output script file
340 platform: Target platform ('windows', 'unix', or None for auto-detect)
342 Raises:
343 FileOperationError: If unable to write to output file
344 """
345 if platform is None:
346 import sys
348 platform = "windows" if sys.platform == "win32" else "unix"
350 output_path = Path(output_file)
352 try:
353 with open(output_path, "w", encoding="utf-8") as f:
354 if platform == "windows":
355 # Generate Batch script for Windows
356 for env_var, config_key in self.ENV_MAPPINGS.items():
357 value = self._config.get(config_key)
358 if value is None:
359 continue
360 f.write(f"set {env_var}={value}\n")
361 else:
362 # Generate Bash script for Unix/Linux/macOS
363 f.write("#!/bin/bash\n")
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 f.write(f"export {env_var}={shlex.quote(str(value))}\n")
369 except OSError as e:
370 raise FileOperationError(
371 f"Could not write to {output_path}: {e}",
372 str(output_path),
373 "export",
374 ) from e
376 # ///////////////////////////////////////////////////////////////
377 # REPRESENTATION METHODS
378 # ///////////////////////////////////////////////////////////////
380 def __str__(self) -> str:
381 """String representation of the configuration."""
382 return f"ConfigurationManager(config_file={self._config_file})"
384 def __repr__(self) -> str:
385 """Detailed string representation of the configuration."""
386 return f"ConfigurationManager(config_file={self._config_file}, config={self._config})"