Coverage for src / ezpl / ezpl.py: 70.49%
269 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-13 19:35 +0000
1# ///////////////////////////////////////////////////////////////
2# EZPL - Main logging singleton
3# Project: ezpl
4# ///////////////////////////////////////////////////////////////
6from __future__ import annotations
8# ///////////////////////////////////////////////////////////////
9# IMPORTS
10# ///////////////////////////////////////////////////////////////
11# Standard library imports
12import sys
13import threading
14import warnings
15from collections.abc import Generator
16from contextlib import contextmanager
17from pathlib import Path
18from typing import Any
19from uuid import uuid4
21# Third-party imports
22from loguru import logger
24# Local imports
25from .config import ConfigurationManager
26from .core.exceptions import EzplError, InitializationError
27from .handlers import EzLogger, EzPrinter
29# ///////////////////////////////////////////////////////////////
30# GLOBALS
31# ///////////////////////////////////////////////////////////////
33APP_PATH = Path(sys.argv[0]).parent
35# ///////////////////////////////////////////////////////////////
36# CLASSES
37# ///////////////////////////////////////////////////////////////
40class Ezpl:
41 """
42 Main logging singleton for the Ezpl framework.
44 Ezpl provides a unified, thread-safe interface for console and file logging
45 with advanced features including indentation management, pattern-based logging,
46 and dynamic progress bars. It implements the Singleton pattern to ensure only
47 one instance exists throughout the application lifecycle.
49 Attributes:
50 _instance: The singleton instance of Ezpl
51 _lock: Thread lock for synchronized access
52 _config_locked: Whether configuration can be modified
53 _log_file: Path to the log file
54 _printer: Console output handler
55 _logger: File logging handler
56 _config_manager: Configuration manager instance
58 Note:
59 Once initialized, Ezpl cannot be re-configured unless reset.
60 Access it via the singleton pattern or module-level functions.
62 Example:
63 >>> log = Ezpl()
64 >>> log.printer.log("INFO", "Application started")
65 >>> log.logger.log("INFO", "Starting logging to file")
66 """
68 _instance: Ezpl | None = None
69 _lock: threading.RLock = threading.RLock()
70 _config_locked: bool = False
71 _config_lock_owner: str | None = None
72 _config_lock_token: str | None = None
73 _log_file: Path
74 _printer: EzPrinter
75 _logger: EzLogger
76 _config_manager: ConfigurationManager
78 # ///////////////////////////////////////////////////////////////
79 # INIT
80 # ///////////////////////////////////////////////////////////////
82 @classmethod
83 def is_initialized(cls) -> bool:
84 """
85 Return True if the Ezpl singleton has already been created.
87 Useful for libraries that want to know whether they are the first
88 to initialize Ezpl or if they should avoid re-configuring it.
89 """
90 return cls._instance is not None
92 def __new__(
93 cls,
94 log_file: Path | str | None = None,
95 log_level: str | None = None,
96 printer_level: str | None = None,
97 file_logger_level: str | None = None,
98 log_rotation: str | None = None,
99 log_retention: str | None = None,
100 log_compression: str | None = None,
101 indent_step: int | None = None,
102 indent_symbol: str | None = None,
103 base_indent_symbol: str | None = None,
104 ) -> Ezpl:
105 """
106 Creates and returns a new instance of Ezpl if none exists.
108 **Notes:**
109 Ensures only one instance of Ezpl exists (Singleton pattern).
111 **Priority order for configuration (for each parameter):**
112 1. Arguments passed directly (highest priority)
113 2. Environment variables (EZPL_*)
114 3. Configuration file (~/.ezpl/config.json)
115 4. Default values (lowest priority)
117 **Args:**
119 * `log_file` (Path | str, optional): Path to the log file
120 * `log_level` (str, optional): Global log level (applies to both printer and logger)
121 * `printer_level` (str, optional): Printer log level
122 * `file_logger_level` (str, optional): File logger level
123 * `log_rotation` (str, optional): Rotation setting (e.g., "10 MB", "1 day")
124 * `log_retention` (str, optional): Retention period (e.g., "7 days")
125 * `log_compression` (str, optional): Compression format (e.g., "zip", "gz")
126 * `indent_step` (int, optional): Indentation step size
127 * `indent_symbol` (str, optional): Symbol for indentation
128 * `base_indent_symbol` (str, optional): Base indentation symbol
130 **Returns:**
132 * `Ezpl`: The singleton instance of the Ezpl class.
134 **Raises:**
136 * `None`.
137 """
139 # //////
140 # Double-checked locking pattern for thread-safe singleton
141 if cls._instance is None:
142 with cls._lock:
143 # Check again after acquiring lock (double-checked locking)
144 if cls._instance is None: 144 ↛ 269line 144 didn't jump to line 269
145 logger.remove()
147 # Initialize configuration manager
148 cls._config_manager = ConfigurationManager()
150 # Determine configuration values with priority: arg > env > config file > default
151 # Helper function to get value with priority order
152 def get_config_value(
153 arg_value, config_key: str, getter_method
154 ) -> Any:
155 """
156 Get configuration value with priority: arg > env > config file > default
158 Args:
159 arg_value: Value from argument (can be None)
160 config_key: Configuration key name
161 getter_method: Method to get default value from config manager
163 Returns:
164 Final configuration value
165 """
166 # Priority 1: Argument direct
167 if arg_value is not None:
168 return arg_value
170 # Priority 2: Environment variable (already loaded in config_manager)
171 # Priority 3: Config file (already loaded in config_manager)
172 # Priority 4: Default (via getter method)
173 # The config_manager already has the correct priority (env > file > default)
174 config_value = cls._config_manager.get(config_key)
175 if config_value is not None: 175 ↛ 179line 175 didn't jump to line 179 because the condition on line 175 was always true
176 return config_value
178 # Fallback to default via getter
179 return getter_method()
181 # Log file
182 if log_file:
183 cls._log_file = Path(log_file)
184 else:
185 cls._log_file = cls._config_manager.get_log_file()
187 # Log level (global)
188 final_log_level = get_config_value(
189 log_level, "log-level", cls._config_manager.get_log_level
190 )
192 # Printer level
193 final_printer_level = get_config_value(
194 printer_level,
195 "printer-level",
196 cls._config_manager.get_printer_level,
197 )
199 # File logger level
200 final_file_logger_level = get_config_value(
201 file_logger_level,
202 "file-logger-level",
203 cls._config_manager.get_file_logger_level,
204 )
206 # Rotation settings (can be None)
207 # Priority: arg > env > config file > default
208 # Note: If arg is None (default), we check env/config/default
209 # If user wants to explicitly set None, they can pass None or use configure()
210 final_rotation = (
211 log_rotation
212 if log_rotation is not None
213 else cls._config_manager.get_log_rotation()
214 )
215 final_retention = (
216 log_retention
217 if log_retention is not None
218 else cls._config_manager.get_log_retention()
219 )
220 final_compression = (
221 log_compression
222 if log_compression is not None
223 else cls._config_manager.get_log_compression()
224 )
226 # Indent settings
227 final_indent_step = get_config_value(
228 indent_step, "indent-step", cls._config_manager.get_indent_step
229 )
230 final_indent_symbol = get_config_value(
231 indent_symbol,
232 "indent-symbol",
233 cls._config_manager.get_indent_symbol,
234 )
235 final_base_indent_symbol = get_config_value(
236 base_indent_symbol,
237 "base-indent-symbol",
238 cls._config_manager.get_base_indent_symbol,
239 )
241 instance = object.__new__(cls)
242 cls._instance = instance
244 # Initialize printer with resolved configuration
245 cls._printer = EzPrinter(
246 level=final_printer_level,
247 indent_step=final_indent_step,
248 indent_symbol=final_indent_symbol,
249 base_indent_symbol=final_base_indent_symbol,
250 )
252 # Initialize logger with resolved configuration
253 cls._logger = EzLogger(
254 log_file=cls._log_file,
255 level=final_file_logger_level,
256 rotation=final_rotation,
257 retention=final_retention,
258 compression=final_compression,
259 )
261 # Apply global log level with priority: specific > global
262 cls._instance._apply_level_priority(
263 printer_level=printer_level,
264 file_logger_level=file_logger_level,
265 global_level=final_log_level,
266 )
268 # Type narrowing: _instance is guaranteed to be set at this point
269 if cls._instance is None: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 raise InitializationError("Singleton initialization failed unexpectedly")
271 return cls._instance
273 # ///////////////////////////////////////////////////////////////
274 # PRIVATE HELPERS
275 # ///////////////////////////////////////////////////////////////
277 def _apply_level_priority(
278 self,
279 *,
280 printer_level: str | None = None,
281 file_logger_level: str | None = None,
282 global_level: str | None = None,
283 force: bool = True,
284 owner: str | None = None,
285 token: str | None = None,
286 ) -> None:
287 """
288 Apply log levels with priority: specific level > global level.
290 Only sets levels when a non-None value can be resolved.
291 If a specific level is not provided, global_level is used as fallback.
292 """
293 effective_printer = printer_level or global_level
294 effective_logger = file_logger_level or global_level
296 if effective_printer:
297 if printer_level and global_level and printer_level != global_level:
298 logger.debug(
299 f"Ezpl: printer_level='{printer_level}' overrides global_level='{global_level}'"
300 )
301 self.set_printer_level(
302 effective_printer,
303 force=force,
304 owner=owner,
305 token=token,
306 )
308 if effective_logger:
309 if file_logger_level and global_level and file_logger_level != global_level:
310 logger.debug(
311 f"Ezpl: file_logger_level='{file_logger_level}' overrides global_level='{global_level}'"
312 )
313 self.set_logger_level(
314 effective_logger,
315 force=force,
316 owner=owner,
317 token=token,
318 )
320 def _rebuild_logger(
321 self,
322 *,
323 level: str | None = None,
324 ) -> None:
325 """
326 Close the current file logger and reinitialize it with current configuration.
328 Args:
329 level: Override the log level. If None, preserves the current level.
330 """
331 current_level = (
332 level
333 or (
334 self._logger.level
335 if hasattr(self, "_logger") and self._logger
336 else None
337 )
338 or self._config_manager.get_file_logger_level()
339 )
340 try:
341 if hasattr(self, "_logger") and self._logger: 341 ↛ 345line 341 didn't jump to line 345 because the condition on line 341 was always true
342 self._logger.close()
343 except (EzplError, OSError, RuntimeError) as e:
344 logger.error(f"Error while closing logger: {e}")
345 self._logger = EzLogger(
346 log_file=self._log_file,
347 level=current_level,
348 rotation=self._config_manager.get_log_rotation(),
349 retention=self._config_manager.get_log_retention(),
350 compression=self._config_manager.get_log_compression(),
351 )
353 def _rebuild_printer(
354 self,
355 *,
356 level: str | None = None,
357 ) -> None:
358 """
359 Reinitialize the console printer with current configuration.
361 Args:
362 level: Override the log level. If None, preserves the current level.
363 """
364 current_level = (
365 level
366 or (
367 self._printer.level
368 if hasattr(self, "_printer") and self._printer
369 else None
370 )
371 or self._config_manager.get_printer_level()
372 )
373 self._printer = EzPrinter(
374 level=current_level,
375 indent_step=self._config_manager.get_indent_step(),
376 indent_symbol=self._config_manager.get_indent_symbol(),
377 base_indent_symbol=self._config_manager.get_base_indent_symbol(),
378 )
380 # ///////////////////////////////////////////////////////////////
381 # GETTER
382 # ///////////////////////////////////////////////////////////////
384 def get_printer(self) -> EzPrinter:
385 """
386 Returns the EzPrinter instance.
388 **Returns:**
390 * EzPrinter: The console printer instance providing info(), debug(), success(), etc.
391 Implements PrinterProtocol for type safety.
392 """
393 return self._printer
395 # ///////////////////////////////////////////////////////////////
397 def get_logger(self) -> EzLogger:
398 """
399 Returns the EzLogger instance.
401 **Returns:**
403 * EzLogger: The file logger instance for file logging.
404 Use logger.info(), logger.debug(), etc. directly.
405 For advanced loguru features, use logger.get_loguru()
406 Implements LoggerProtocol for type safety.
407 """
408 return self._logger
410 # ///////////////////////////////////////////////////////////////
411 # UTILS METHODS
412 # ///////////////////////////////////////////////////////////////
414 @property
415 def printer_level(self) -> str:
416 """Return the current printer logging level."""
417 return self._printer.level
419 @property
420 def logger_level(self) -> str:
421 """Return the current file logger logging level."""
422 return self._logger.level
424 def set_level(
425 self,
426 level: str,
427 *,
428 force: bool = False,
429 owner: str | None = None,
430 token: str | None = None,
431 ) -> None:
432 """
433 Set the log level for both the printer and the logger simultaneously (compatibility method).
435 **Args:**
437 * `level` (str): The desired log level (e.g., "INFO", "WARNING").
438 * `force` (bool): If True, bypasses the configuration lock.
440 **Returns:**
442 * `None`.
443 """
444 if not self._can_write_config(force=force, owner=owner, token=token): 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true
445 warnings.warn(
446 "Ezpl configuration is locked. Call Ezpl.unlock_config() or pass a "
447 "valid owner/token with force=True to override.",
448 UserWarning,
449 stacklevel=2,
450 )
451 return
452 self.set_logger_level(level, force=force, owner=owner, token=token)
453 self.set_printer_level(level, force=force, owner=owner, token=token)
455 def set_printer_level(
456 self,
457 level: str,
458 *,
459 force: bool = False,
460 owner: str | None = None,
461 token: str | None = None,
462 ) -> None:
463 """
464 Set the log level for the printer only.
466 **Args:**
468 * `level` (str): The desired log level for the printer.
469 * `force` (bool): If True, bypasses the configuration lock.
471 **Returns:**
473 * `None`.
474 """
475 if not self._can_write_config(force=force, owner=owner, token=token): 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true
476 warnings.warn(
477 "Ezpl configuration is locked. Call Ezpl.unlock_config() or pass a "
478 "valid owner/token with force=True to override.",
479 UserWarning,
480 stacklevel=2,
481 )
482 return
483 self._printer.set_level(level)
485 def set_logger_level(
486 self,
487 level: str,
488 *,
489 force: bool = False,
490 owner: str | None = None,
491 token: str | None = None,
492 ) -> None:
493 """
494 Set the log level for the file logger only.
496 **Args:**
498 * `level` (str): The desired log level for the file logger.
499 * `force` (bool): If True, bypasses the configuration lock.
501 **Returns:**
503 * `None`.
504 """
505 if not self._can_write_config(force=force, owner=owner, token=token): 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 warnings.warn(
507 "Ezpl configuration is locked. Call Ezpl.unlock_config() or pass a "
508 "valid owner/token with force=True to override.",
509 UserWarning,
510 stacklevel=2,
511 )
512 return
513 self._logger.set_level(level)
515 # ///////////////////////////////////////////////////////////////
517 def add_separator(self) -> None:
518 """
519 Adds a separator to the log file.
521 **Returns:**
523 * `None`.
524 """
525 self._logger.add_separator()
527 # ///////////////////////////////////////////////////////////////
529 @contextmanager
530 def manage_indent(self) -> Generator[None, None, None]:
531 """
532 Context manager to manage indentation level.
534 **Returns:**
536 * `None`.
537 """
538 with self._printer.manage_indent():
539 yield
541 # ///////////////////////////////////////////////////////////////
542 # ENHANCED METHODS
543 # ///////////////////////////////////////////////////////////////
545 @classmethod
546 def reset(cls) -> None:
547 """
548 Reset the singleton instance (useful for testing).
550 Warning: This will destroy the current instance and all its state.
551 """
552 if cls._instance is not None:
553 # Close logger handlers to release file handles (important on Windows)
554 try:
555 if hasattr(cls._instance, "_logger") and cls._instance._logger: 555 ↛ 559line 555 didn't jump to line 559 because the condition on line 555 was always true
556 cls._instance._logger.close()
557 except (EzplError, OSError, RuntimeError) as e:
558 logger.error(f"Error during cleanup: {e}")
559 cls._instance = None
560 # Also reset configuration lock
561 cls._config_locked = False
562 cls._config_lock_owner = None
563 cls._config_lock_token = None
565 # ------------------------------------------------
566 # CONFIG LOCK CONTROL
567 # ------------------------------------------------
569 @classmethod
570 def lock_config(cls, owner: str = "app") -> str | None:
571 """
572 Lock Ezpl configuration so that future configure() calls are ignored
573 unless explicitly forced.
575 Intended usage:
576 1. Root application configures Ezpl once
577 2. Calls Ezpl.lock_config()
578 3. Libraries calling configure() later will not override settings
579 """
580 with cls._lock:
581 normalized_owner = owner.strip() if owner.strip() else "app"
583 if cls._config_locked and cls._config_lock_owner not in (
584 None,
585 normalized_owner,
586 ):
587 warnings.warn(
588 f"Ezpl configuration is already locked by '{cls._config_lock_owner}'.",
589 UserWarning,
590 stacklevel=2,
591 )
592 return None
594 cls._config_locked = True
595 cls._config_lock_owner = normalized_owner
596 cls._config_lock_token = uuid4().hex
597 return cls._config_lock_token
599 @classmethod
600 def unlock_config(
601 cls,
602 *,
603 owner: str | None = None,
604 token: str | None = None,
605 force: bool = False,
606 ) -> bool:
607 """
608 Unlock Ezpl configuration.
610 Use with care: this allows configure() to change global logging
611 configuration again.
612 """
613 with cls._lock:
614 if not cls._config_locked: 614 ↛ 615line 614 didn't jump to line 615 because the condition on line 614 was never true
615 return True
617 owner_match = owner is not None and owner == cls._config_lock_owner
618 token_match = token is not None and token == cls._config_lock_token
620 if not force and not owner_match and not token_match:
621 warnings.warn(
622 "Unlock denied: provide matching owner or token, or use force=True.",
623 UserWarning,
624 stacklevel=2,
625 )
626 return False
628 cls._config_locked = False
629 cls._config_lock_owner = None
630 cls._config_lock_token = None
631 return True
633 @classmethod
634 def config_lock_info(cls) -> dict[str, Any]:
635 """Return current configuration lock state for diagnostics."""
636 return {
637 "locked": cls._config_locked,
638 "owner": cls._config_lock_owner,
639 "has_token": cls._config_lock_token is not None,
640 }
642 @classmethod
643 def _can_write_config(
644 cls,
645 *,
646 force: bool = False,
647 owner: str | None = None,
648 token: str | None = None,
649 ) -> bool:
650 """Return True when a configuration write is authorized under lock rules."""
651 if not cls._config_locked:
652 return True
654 if not force: 654 ↛ 655line 654 didn't jump to line 655 because the condition on line 654 was never true
655 return False
657 # Backward compatibility: if no owner/token metadata exists, force still works.
658 if cls._config_lock_owner is None and cls._config_lock_token is None: 658 ↛ 659line 658 didn't jump to line 659 because the condition on line 658 was never true
659 return True
661 owner_match = owner is not None and owner == cls._config_lock_owner
662 token_match = token is not None and token == cls._config_lock_token
663 return owner_match or token_match
665 def set_log_file(self, log_file: Path | str) -> None:
666 """
667 Change the log file (requires reinitialization of the logger).
669 Args:
670 log_file: New path to the log file
672 Note: This will reinitialize the file logger but keep the singleton instance.
673 """
674 new_log_file = Path(log_file)
675 if new_log_file != self._log_file: 675 ↛ exitline 675 didn't return from function 'set_log_file' because the condition on line 675 was always true
676 self._log_file = new_log_file
677 # Update configuration
678 self._config_manager.set("log-file", str(new_log_file))
679 self._rebuild_logger()
681 def get_log_file(self) -> Path:
682 """
683 Get the current log file path.
685 Returns:
686 Path to the current log file
687 """
688 return self._log_file
690 def get_config(self) -> ConfigurationManager:
691 """
692 Get the current configuration manager.
694 Returns:
695 ConfigurationManager instance for accessing and modifying configuration
696 """
697 return self._config_manager
699 # ///////////////////////////////////////////////////////////////
700 # HANDLER OVERRIDE METHODS
701 # ///////////////////////////////////////////////////////////////
703 def set_printer_class(
704 self,
705 printer_class: type[EzPrinter] | EzPrinter,
706 **init_kwargs: Any,
707 ) -> None:
708 """
709 Replace the current printer with a custom printer class or instance.
711 Allows users to override the default printer with a custom class that
712 inherits from EzPrinter. The method preserves
713 current configuration values (level, indentation settings) unless
714 explicitly overridden in init_kwargs.
716 Args:
717 printer_class: Custom printer class inheriting from EzPrinter,
718 or an already instantiated EzPrinter instance
719 **init_kwargs: Optional initialization parameters for the printer
720 class. If not provided, current configuration values are used.
722 Raises:
723 TypeError: If printer_class is not a valid class or instance
724 ValidationError: If initialization parameters are invalid
726 Example:
727 >>> from ezpl import Ezpl, EzPrinter
728 >>>
729 >>> class CustomPrinter(EzPrinter):
730 ... def info(self, message):
731 ... super().info(f"[CUSTOM] {message}")
732 >>>
733 >>> ezpl = Ezpl()
734 >>> ezpl.set_printer_class(CustomPrinter, level="DEBUG")
735 >>> ezpl.get_printer().info("Test")
736 [CUSTOM] Test
737 """
738 from .core.exceptions import ValidationError
740 # If it's already an instance, use it directly
741 if isinstance(printer_class, EzPrinter):
742 new_printer = printer_class
743 # If it's a class, instantiate it
744 elif isinstance(printer_class, type):
745 # Validate that it's a subclass of EzPrinter
746 if not issubclass(printer_class, EzPrinter):
747 raise TypeError(
748 f"printer_class must be a subclass of {EzPrinter.__name__}, "
749 f"got {printer_class.__name__}"
750 )
752 # Preserve current configuration values if not provided
753 current_level = (
754 self._printer.level
755 if hasattr(self, "_printer") and self._printer
756 else self._config_manager.get_printer_level()
757 )
758 current_indent_step = (
759 self._printer.indent_step
760 if hasattr(self, "_printer") and self._printer
761 else self._config_manager.get_indent_step()
762 )
763 current_indent_symbol = (
764 self._printer.indent_symbol
765 if hasattr(self, "_printer") and self._printer
766 else self._config_manager.get_indent_symbol()
767 )
768 current_base_indent_symbol = (
769 self._printer.base_indent_symbol
770 if hasattr(self, "_printer") and self._printer
771 else self._config_manager.get_base_indent_symbol()
772 )
774 # Merge kwargs with default values
775 init_params = {
776 "level": init_kwargs.pop("level", current_level),
777 "indent_step": init_kwargs.pop("indent_step", current_indent_step),
778 "indent_symbol": init_kwargs.pop(
779 "indent_symbol", current_indent_symbol
780 ),
781 "base_indent_symbol": init_kwargs.pop(
782 "base_indent_symbol", current_base_indent_symbol
783 ),
784 }
785 init_params.update(init_kwargs)
787 # Create new instance
788 try:
789 new_printer = printer_class(**init_params)
790 except Exception as e:
791 raise ValidationError(
792 f"Failed to initialize printer class {printer_class.__name__}: {e}",
793 "printer_class",
794 str(printer_class),
795 ) from e
796 else:
797 raise TypeError(
798 f"printer_class must be a class or an instance of {EzPrinter.__name__}, "
799 f"got {type(printer_class).__name__}"
800 )
802 # Replace the instance
803 self._printer = new_printer
805 def set_logger_class(
806 self,
807 logger_class: type[EzLogger] | EzLogger,
808 **init_kwargs: Any,
809 ) -> None:
810 """
811 Replace the current logger with a custom logger class or instance.
813 Allows users to override the default logger with a custom class that
814 inherits from EzLogger. The method preserves current
815 configuration values (level, rotation, retention, compression) unless
816 explicitly overridden in init_kwargs.
818 Args:
819 logger_class: Custom logger class inheriting from EzLogger,
820 or an already instantiated EzLogger instance
821 **init_kwargs: Optional initialization parameters for the logger
822 class. If not provided, current configuration values are used.
824 Raises:
825 TypeError: If logger_class is not a valid class or instance
826 ValidationError: If initialization parameters are invalid
827 FileOperationError: If file operations fail during logger creation
828 (may be raised by the logger class constructor)
830 Example:
831 >>> from ezpl import Ezpl, EzLogger
832 >>>
833 >>> class CustomLogger(EzLogger):
834 ... def info(self, message):
835 ... super().info(f"[CUSTOM LOG] {message}")
836 >>>
837 >>> ezpl = Ezpl()
838 >>> ezpl.set_logger_class(CustomLogger, log_file="custom.log")
839 >>> ezpl.get_logger().info("Test")
840 """
841 from .core.exceptions import ValidationError
843 # If it's already an instance, use it directly
844 if isinstance(logger_class, EzLogger):
845 new_logger = logger_class
846 # If it's a class, instantiate it
847 elif isinstance(logger_class, type):
848 # Validate that it's a subclass of EzLogger
849 if not issubclass(logger_class, EzLogger):
850 raise TypeError(
851 f"logger_class must be a subclass of {EzLogger.__name__}, "
852 f"got {logger_class.__name__}"
853 )
855 # Preserve current configuration values if not provided
856 current_level = (
857 self._logger.level
858 if hasattr(self, "_logger") and self._logger
859 else self._config_manager.get_file_logger_level()
860 )
861 current_log_file = (
862 self._log_file
863 if hasattr(self, "_log_file")
864 else self._config_manager.get_log_file()
865 )
866 current_rotation = (
867 self._logger.rotation
868 if hasattr(self, "_logger") and self._logger
869 else self._config_manager.get_log_rotation()
870 )
871 current_retention = (
872 self._logger.retention
873 if hasattr(self, "_logger") and self._logger
874 else self._config_manager.get_log_retention()
875 )
876 current_compression = (
877 self._logger.compression
878 if hasattr(self, "_logger") and self._logger
879 else self._config_manager.get_log_compression()
880 )
882 # Merge kwargs with default values
883 init_params = {
884 "log_file": init_kwargs.pop("log_file", current_log_file),
885 "level": init_kwargs.pop("level", current_level),
886 "rotation": init_kwargs.pop("rotation", current_rotation),
887 "retention": init_kwargs.pop("retention", current_retention),
888 "compression": init_kwargs.pop("compression", current_compression),
889 }
890 init_params.update(init_kwargs)
892 # Close previous logger before creating new one to avoid resource leaks
893 try:
894 if hasattr(self, "_logger") and self._logger:
895 self._logger.close()
896 except (EzplError, OSError, RuntimeError) as e:
897 logger.error(f"Error while closing previous logger: {e}")
899 # Create new instance
900 try:
901 new_logger = logger_class(**init_params)
902 except Exception as e:
903 raise ValidationError(
904 f"Failed to initialize logger class {logger_class.__name__}: {e}",
905 "logger_class",
906 str(logger_class),
907 ) from e
908 else:
909 raise TypeError(
910 f"logger_class must be a class or an instance of {EzLogger.__name__}, "
911 f"got {type(logger_class).__name__}"
912 )
914 # Replace the instance
915 self._logger = new_logger
917 # ///////////////////////////////////////////////////////////////
918 # CONFIGURATION METHODS
919 # ///////////////////////////////////////////////////////////////
921 def reload_config(self) -> None:
922 """
923 Reload configuration from file and environment variables.
925 This method reloads the configuration and reapplies it to handlers.
926 Useful when environment variables or the config file have changed
927 after the singleton was initialized.
929 Note: This will reinitialize handlers with the new configuration.
930 """
931 # Reload configuration
932 self._config_manager.reload()
934 # Get configuration values
935 printer_level = self._config_manager.get_printer_level()
936 file_logger_level = self._config_manager.get_file_logger_level()
937 global_log_level = self._config_manager.get_log_level()
939 # Check if specific levels are explicitly set (not just defaults)
940 # Priority: specific levels > global level
941 # Only apply global level if specific levels are not explicitly set
942 printer_level_explicit = self._config_manager.has_key("printer-level")
943 file_logger_level_explicit = self._config_manager.has_key("file-logger-level")
944 global_log_level_explicit = self._config_manager.has_key("log-level")
946 # Respect manually set levels: don't override if set via set_level()
947 printer_manually_set = self._printer.level_manually_set
948 logger_manually_set = self._logger.level_manually_set
950 # Reapply to handlers with priority logic (skip if manually set)
951 if not printer_manually_set: 951 ↛ 952line 951 didn't jump to line 952 because the condition on line 951 was never true
952 effective_printer = (
953 printer_level
954 if printer_level_explicit
955 else global_log_level
956 if global_log_level_explicit
957 else printer_level
958 )
959 self.set_printer_level(effective_printer, force=True)
960 self._printer.mark_level_as_configured()
962 if not logger_manually_set: 962 ↛ 963line 962 didn't jump to line 963 because the condition on line 962 was never true
963 effective_logger = (
964 file_logger_level
965 if file_logger_level_explicit
966 else (
967 global_log_level if global_log_level_explicit else file_logger_level
968 )
969 )
970 self.set_logger_level(effective_logger, force=True)
971 self._logger.mark_level_as_configured()
973 # Reinitialize logger with new rotation / retention / compression settings
974 self._rebuild_logger()
975 if not logger_manually_set: 975 ↛ 976line 975 didn't jump to line 976 because the condition on line 975 was never true
976 self._logger.mark_level_as_configured()
978 # Reinitialize printer with new indent settings
979 self._rebuild_printer()
980 if not printer_manually_set: 980 ↛ 981line 980 didn't jump to line 981 because the condition on line 980 was never true
981 self._printer.mark_level_as_configured()
983 def configure(
984 self, config_dict: dict[str, Any] | None = None, **kwargs: Any
985 ) -> bool:
986 """
987 Configure Ezpl dynamically.
989 Args:
990 config_dict: Dictionary of configuration values to update
991 **kwargs: Configuration options (alternative to config_dict):
992 - log_file or log-file: Path to log file
993 - printer_level or printer-level: Printer log level
994 - logger_level or file-logger-level: File logger level
995 - level or log-level: Set both printer and logger level
996 - log_rotation or log-rotation: Rotation setting (e.g., "10 MB", "1 day")
997 - log_retention or log-retention: Retention period (e.g., "7 days")
998 - log_compression or log-compression: Compression format (e.g., "zip", "gz")
999 - indent_step or indent-step: Indentation step size
1000 - indent_symbol or indent-symbol: Symbol for indentation
1001 - base_indent_symbol or base-indent-symbol: Base indentation symbol
1003 Returns:
1004 True if configuration was applied, False if it was blocked by lock.
1006 Note: Changes are persisted to the configuration file.
1007 """
1008 # Merge config_dict and kwargs
1009 if config_dict:
1010 kwargs.update(config_dict)
1012 # Special control flag (not stored in configuration):
1013 # - force=True allows configure() even when configuration is locked
1014 force = kwargs.pop("force", False)
1015 owner = kwargs.pop("owner", None)
1016 token = kwargs.pop("token", None)
1018 # If configuration is locked and not forced, warn and return False
1019 if not self._can_write_config(force=force, owner=owner, token=token):
1020 warnings.warn(
1021 "Ezpl configuration is locked. Call Ezpl.unlock_config() or pass a "
1022 "valid owner/token with force=True to override.",
1023 UserWarning,
1024 stacklevel=2,
1025 )
1026 return False
1028 # Normalize keys: convert underscores to hyphens for consistency
1029 normalized_config = {}
1030 key_mapping = {
1031 "log_file": "log-file",
1032 "printer_level": "printer-level",
1033 "logger_level": "file-logger-level",
1034 "level": "log-level",
1035 "log_rotation": "log-rotation",
1036 "log_retention": "log-retention",
1037 "log_compression": "log-compression",
1038 "indent_step": "indent-step",
1039 "indent_symbol": "indent-symbol",
1040 "base_indent_symbol": "base-indent-symbol",
1041 }
1043 for key, value in kwargs.items():
1044 # Use normalized key if mapping exists, otherwise keep original
1045 normalized_key = key_mapping.get(key, key)
1046 normalized_config[normalized_key] = value
1048 # Update configuration manager
1049 self._config_manager.update(normalized_config)
1050 self._config_manager.save()
1052 # Apply changes to handlers
1053 if "log-file" in normalized_config: 1053 ↛ 1054line 1053 didn't jump to line 1054 because the condition on line 1053 was never true
1054 self.set_log_file(normalized_config["log-file"])
1056 # Handle log level changes with priority: specific > global
1057 self._apply_level_priority(
1058 printer_level=normalized_config.get("printer-level"),
1059 file_logger_level=normalized_config.get("file-logger-level"),
1060 global_level=normalized_config.get("log-level"),
1061 force=force,
1062 owner=owner,
1063 token=token,
1064 )
1066 # Reinitialize logger if rotation settings changed
1067 rotation_changed = any(
1068 key in normalized_config
1069 for key in ["log-rotation", "log-retention", "log-compression"]
1070 )
1071 if rotation_changed:
1072 self._rebuild_logger()
1074 # Reinitialize printer if indent settings changed
1075 indent_changed = any(
1076 key in normalized_config
1077 for key in ["indent-step", "indent-symbol", "base-indent-symbol"]
1078 )
1079 if indent_changed: 1079 ↛ 1080line 1079 didn't jump to line 1080 because the condition on line 1079 was never true
1080 self._rebuild_printer()
1082 return True