Coverage for src / ezplog / ezpl.py: 74.12%
308 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-03 16:27 +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 .app_mode import InterceptHandler
26from .config import ConfigurationManager
27from .core.exceptions import EzplError, InitializationError
28from .handlers import EzLogger, EzPrinter
30# ///////////////////////////////////////////////////////////////
31# GLOBALS
32# ///////////////////////////////////////////////////////////////
34_APP_PATH = Path(sys.argv[0]).parent
36# ///////////////////////////////////////////////////////////////
37# CLASSES
38# ///////////////////////////////////////////////////////////////
41class Ezpl:
42 """
43 Main logging singleton for the Ezpl framework.
45 Ezpl provides a unified, thread-safe interface for console and file logging
46 with advanced features including indentation management, pattern-based logging,
47 and dynamic progress bars. It implements the Singleton pattern to ensure only
48 one instance exists throughout the application lifecycle.
50 Attributes:
51 _instance: The singleton instance of Ezpl
52 _lock: Thread lock for synchronized access
53 _config_locked: Whether configuration can be modified
54 _log_file: Path to the log file
55 _printer: Console output handler
56 _logger: File logging handler
57 _config_manager: Configuration manager instance
59 Note:
60 Once initialized, Ezpl cannot be re-configured unless reset.
61 Access it via the singleton pattern or module-level functions.
63 Example:
64 >>> log = Ezpl()
65 >>> log.printer.log("INFO", "Application started")
66 >>> log.logger.log("INFO", "Starting logging to file")
67 """
69 _instance: Ezpl | None = None
70 _lock: threading.RLock = threading.RLock()
71 _config_locked: bool = False
72 _config_lock_token: str | None = None
73 _log_file: Path
74 _printer: EzPrinter
75 _logger: EzLogger
76 _config_manager: ConfigurationManager
77 _lib_printer_hook_enabled: bool = True
78 _stdlib_hook_targets: set[str | None] = set()
80 # ///////////////////////////////////////////////////////////////
81 # INIT
82 # ///////////////////////////////////////////////////////////////
84 @classmethod
85 def is_initialized(cls) -> bool:
86 """
87 Return True if the Ezpl singleton has already been created.
89 Useful for libraries that want to know whether they are the first
90 to initialize Ezpl or if they should avoid re-configuring it.
91 """
92 return cls._instance is not None
94 def __new__(
95 cls,
96 log_file: Path | str | None = None,
97 log_level: str | None = None,
98 printer_level: str | None = None,
99 file_logger_level: str | None = None,
100 log_rotation: str | None = None,
101 log_retention: str | None = None,
102 log_compression: str | None = None,
103 indent_step: int | None = None,
104 indent_symbol: str | None = None,
105 base_indent_symbol: str | None = None,
106 *,
107 lock_config: bool = False,
108 hook_logger: bool = True,
109 hook_printer: bool = True,
110 logger_names: list[str] | None = None,
111 ) -> Ezpl:
112 """
113 Creates and returns a new instance of Ezpl if none exists.
115 **Notes:**
116 Ensures only one instance of Ezpl exists (Singleton pattern).
118 **Priority order for configuration (for each parameter):**
119 1. Arguments passed directly (highest priority)
120 2. Environment variables (EZPL_*)
121 3. Configuration file (~/.ezpl/config.json)
122 4. Default values (lowest priority)
124 **Args:**
126 * `log_file` (Path | str, optional): Path to the log file
127 * `log_level` (str, optional): Global log level (applies to both printer and logger)
128 * `printer_level` (str, optional): Printer log level
129 * `file_logger_level` (str, optional): File logger level
130 * `log_rotation` (str, optional): Rotation setting (e.g., "10 MB", "1 day")
131 * `log_retention` (str, optional): Retention period (e.g., "7 days")
132 * `log_compression` (str, optional): Compression format (e.g., "zip", "gz")
133 * `indent_step` (int, optional): Indentation step size
134 * `indent_symbol` (str, optional): Symbol for indentation
135 * `base_indent_symbol` (str, optional): Base indentation symbol
136 * `lock_config` (bool): If True, lock configuration immediately after init.
137 Subsequent configure() calls will be blocked unless a valid token is used.
138 The lock token is stored in Ezpl._config_lock_token.
139 * `hook_logger` (bool): If True, install stdlib logging bridge(s) so
140 classic loggers are forwarded to the loguru pipeline.
141 * `hook_printer` (bool): If True, allow ezpl.lib_mode.get_printer()
142 to delegate to the real EzPrinter instance once initialized.
143 * `logger_names` (list[str] | None): Optional stdlib logger names to
144 bridge explicitly. When omitted, the root logger is used.
146 **Returns:**
148 * `Ezpl`: The singleton instance of the Ezpl class.
150 **Raises:**
152 * `None`.
153 """
155 # //////
156 # Double-checked locking pattern for thread-safe singleton
157 if cls._instance is None:
158 with cls._lock:
159 # Check again after acquiring lock (double-checked locking)
160 if cls._instance is None:
161 # Initialize configuration manager
162 cls._config_manager = ConfigurationManager()
164 # Determine configuration values with priority: arg > env > config file > default
165 # Helper function to get value with priority order
166 def get_config_value(
167 arg_value, config_key: str, getter_method
168 ) -> Any:
169 """
170 Get configuration value with priority: arg > env > config file > default
172 Args:
173 arg_value: Value from argument (can be None)
174 config_key: Configuration key name
175 getter_method: Method to get default value from config manager
177 Returns:
178 Final configuration value
179 """
180 # Priority 1: Argument direct
181 if arg_value is not None:
182 return arg_value
184 # Priority 2: Environment variable (already loaded in config_manager)
185 # Priority 3: Config file (already loaded in config_manager)
186 # Priority 4: Default (via getter method)
187 # The config_manager already has the correct priority (env > file > default)
188 config_value = cls._config_manager.get(config_key)
189 if config_value is not None: 189 ↛ 193line 189 didn't jump to line 193 because the condition on line 189 was always true
190 return config_value
192 # Fallback to default via getter
193 return getter_method()
195 # Log file
196 if log_file:
197 cls._log_file = Path(log_file)
198 else:
199 cls._log_file = cls._config_manager.get_log_file()
201 # Log level (global)
202 final_log_level = get_config_value(
203 log_level, "log-level", cls._config_manager.get_log_level
204 )
206 # Printer level
207 final_printer_level = get_config_value(
208 printer_level,
209 "printer-level",
210 cls._config_manager.get_printer_level,
211 )
213 # File logger level
214 final_file_logger_level = get_config_value(
215 file_logger_level,
216 "file-logger-level",
217 cls._config_manager.get_file_logger_level,
218 )
220 # Rotation settings (can be None)
221 # Priority: arg > env > config file > default
222 # Note: If arg is None (default), we check env/config/default
223 # If user wants to explicitly set None, they can pass None or use configure()
224 final_rotation = (
225 log_rotation
226 if log_rotation is not None
227 else cls._config_manager.get_log_rotation()
228 )
229 final_retention = (
230 log_retention
231 if log_retention is not None
232 else cls._config_manager.get_log_retention()
233 )
234 final_compression = (
235 log_compression
236 if log_compression is not None
237 else cls._config_manager.get_log_compression()
238 )
240 # Indent settings
241 final_indent_step = get_config_value(
242 indent_step, "indent-step", cls._config_manager.get_indent_step
243 )
244 final_indent_symbol = get_config_value(
245 indent_symbol,
246 "indent-symbol",
247 cls._config_manager.get_indent_symbol,
248 )
249 final_base_indent_symbol = get_config_value(
250 base_indent_symbol,
251 "base-indent-symbol",
252 cls._config_manager.get_base_indent_symbol,
253 )
255 instance = object.__new__(cls)
256 cls._instance = instance
258 # Initialize printer with resolved configuration
259 cls._printer = EzPrinter(
260 level=final_printer_level,
261 indent_step=final_indent_step,
262 indent_symbol=final_indent_symbol,
263 base_indent_symbol=final_base_indent_symbol,
264 )
266 # Initialize logger with resolved configuration
267 cls._logger = EzLogger(
268 log_file=cls._log_file,
269 level=final_file_logger_level,
270 rotation=final_rotation,
271 retention=final_retention,
272 compression=final_compression,
273 )
275 # Apply global log level with priority: specific > global
276 cls._instance._apply_level_priority(
277 printer_level=printer_level,
278 file_logger_level=file_logger_level,
279 global_level=final_log_level,
280 )
282 # Apply post-init options — inside the critical section,
283 # first construction only (double-checked locking guarantees this)
284 if lock_config: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 cls._config_lock_token = cls.lock_config()
287 cls._set_compatibility_hooks(
288 hook_logger=hook_logger,
289 hook_printer=hook_printer,
290 logger_names=logger_names,
291 )
293 # Type narrowing: _instance is guaranteed to be set at this point
294 if cls._instance is None: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true
295 raise InitializationError("Singleton initialization failed unexpectedly")
296 return cls._instance
298 # ///////////////////////////////////////////////////////////////
299 # PRIVATE HELPERS
300 # ///////////////////////////////////////////////////////////////
302 def _apply_level_priority(
303 self,
304 *,
305 printer_level: str | None = None,
306 file_logger_level: str | None = None,
307 global_level: str | None = None,
308 ) -> None:
309 """
310 Apply log levels with priority: specific level > global level.
312 Only sets levels when a non-None value can be resolved.
313 If a specific level is not provided, global_level is used as fallback.
314 """
315 effective_printer = printer_level if printer_level is not None else global_level
316 effective_logger = (
317 file_logger_level if file_logger_level is not None else global_level
318 )
320 if effective_printer:
321 self.set_printer_level(effective_printer)
323 if effective_logger:
324 self.set_logger_level(effective_logger)
326 def _rebuild_logger(
327 self,
328 *,
329 level: str | None = None,
330 ) -> None:
331 """
332 Close the current file logger and reinitialize it with current configuration.
334 Args:
335 level: Override the log level. If None, preserves the current level.
336 """
337 current_level = (
338 level
339 or (
340 self._logger.level
341 if hasattr(self, "_logger") and self._logger
342 else None
343 )
344 or self._config_manager.get_file_logger_level()
345 )
346 try:
347 if hasattr(self, "_logger") and self._logger: 347 ↛ 351line 347 didn't jump to line 351 because the condition on line 347 was always true
348 self._logger.close()
349 except (EzplError, OSError, RuntimeError) as e:
350 logger.error(f"Error while closing logger: {e}")
351 self._logger = EzLogger(
352 log_file=self._log_file,
353 level=current_level,
354 rotation=self._config_manager.get_log_rotation(),
355 retention=self._config_manager.get_log_retention(),
356 compression=self._config_manager.get_log_compression(),
357 )
359 def _rebuild_printer(
360 self,
361 *,
362 level: str | None = None,
363 ) -> None:
364 """
365 Reinitialize the console printer with current configuration.
367 Args:
368 level: Override the log level. If None, preserves the current level.
369 """
370 current_level = (
371 level
372 or (
373 self._printer.level
374 if hasattr(self, "_printer") and self._printer
375 else None
376 )
377 or self._config_manager.get_printer_level()
378 )
379 self._printer = EzPrinter(
380 level=current_level,
381 indent_step=self._config_manager.get_indent_step(),
382 indent_symbol=self._config_manager.get_indent_symbol(),
383 base_indent_symbol=self._config_manager.get_base_indent_symbol(),
384 )
386 # ///////////////////////////////////////////////////////////////
387 # GETTER
388 # ///////////////////////////////////////////////////////////////
390 def get_printer(self) -> EzPrinter:
391 """
392 Returns the EzPrinter instance.
394 **Returns:**
396 * EzPrinter: The console printer instance providing info(), debug(), success(), etc.
397 Implements PrinterProtocol for type safety.
398 """
399 return self._printer
401 # ///////////////////////////////////////////////////////////////
403 def get_logger(self) -> EzLogger:
404 """
405 Returns the EzLogger instance.
407 **Returns:**
409 * EzLogger: The file logger instance for file logging.
410 Use logger.info(), logger.debug(), etc. directly.
411 For advanced loguru features, use logger.get_loguru()
412 Implements LoggerProtocol for type safety.
413 """
414 return self._logger
416 # ///////////////////////////////////////////////////////////////
417 # UTILS METHODS
418 # ///////////////////////////////////////////////////////////////
420 @property
421 def printer_level(self) -> str:
422 """Return the current printer logging level."""
423 return self._printer.level
425 @property
426 def logger_level(self) -> str:
427 """Return the current file logger logging level."""
428 return self._logger.level
430 def set_level(self, level: str) -> None:
431 """
432 Set the log level for both the printer and the logger simultaneously.
434 **Args:**
436 * `level` (str): The desired log level (e.g., "INFO", "WARNING").
438 **Returns:**
440 * `None`.
441 """
442 if not self._can_write_config(): 442 ↛ 443line 442 didn't jump to line 443 because the condition on line 442 was never true
443 warnings.warn(
444 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock.",
445 UserWarning,
446 stacklevel=2,
447 )
448 return
449 self.set_logger_level(level)
450 self.set_printer_level(level)
452 def set_printer_level(self, level: str) -> None:
453 """
454 Set the log level for the printer only.
456 **Args:**
458 * `level` (str): The desired log level for the printer.
460 **Returns:**
462 * `None`.
463 """
464 if not self._can_write_config(): 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 warnings.warn(
466 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock.",
467 UserWarning,
468 stacklevel=2,
469 )
470 return
471 self._printer.set_level(level)
473 def set_logger_level(self, level: str) -> None:
474 """
475 Set the log level for the file logger only.
477 **Args:**
479 * `level` (str): The desired log level for the file logger.
481 **Returns:**
483 * `None`.
484 """
485 if not self._can_write_config(): 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 warnings.warn(
487 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock.",
488 UserWarning,
489 stacklevel=2,
490 )
491 return
492 self._logger.set_level(level)
494 # ///////////////////////////////////////////////////////////////
495 # FACADE METHODS
496 # ///////////////////////////////////////////////////////////////
498 def debug(self, message: Any) -> None:
499 """Log a debug message to the console printer."""
500 self._printer.debug(message)
502 def info(self, message: Any) -> None:
503 """Log an info message to the console printer."""
504 self._printer.info(message)
506 def success(self, message: Any) -> None:
507 """Log a success message to the console printer."""
508 self._printer.success(message)
510 def warning(self, message: Any) -> None:
511 """Log a warning message to the console printer."""
512 self._printer.warning(message)
514 def error(self, message: Any) -> None:
515 """Log an error message to the console printer."""
516 self._printer.error(message)
518 def critical(self, message: Any) -> None:
519 """Log a critical message to the console printer."""
520 self._printer.critical(message)
522 # ///////////////////////////////////////////////////////////////
524 def add_separator(self) -> None:
525 """
526 Adds a separator to the log file.
528 **Returns:**
530 * `None`.
531 """
532 self._logger.add_separator()
534 # ///////////////////////////////////////////////////////////////
536 @contextmanager
537 def manage_indent(self) -> Generator[None, None, None]:
538 """
539 Context manager to manage indentation level.
541 **Returns:**
543 * `None`.
544 """
545 with self._printer.manage_indent():
546 yield
548 # ///////////////////////////////////////////////////////////////
549 # ENHANCED METHODS
550 # ///////////////////////////////////////////////////////////////
552 @classmethod
553 def reset(cls) -> None:
554 """
555 Reset the singleton instance (useful for testing).
557 Warning: This will destroy the current instance and all its state.
558 """
559 if cls._instance is not None:
560 # Close logger handlers to release file handles (important on Windows)
561 try:
562 if hasattr(cls._instance, "_logger") and cls._instance._logger: 562 ↛ 566line 562 didn't jump to line 566 because the condition on line 562 was always true
563 cls._instance._logger.close()
564 except (EzplError, OSError, RuntimeError) as e:
565 logger.error(f"Error during cleanup: {e}")
566 cls._instance = None
567 cls._remove_all_intercept_handlers()
568 cls._lib_printer_hook_enabled = True
569 # Also reset configuration lock
570 cls._config_locked = False
571 cls._config_lock_token = None
573 def set_compatibility_hooks(
574 self,
575 *,
576 hook_logger: bool = True,
577 hook_printer: bool = True,
578 logger_names: list[str] | None = None,
579 ) -> None:
580 """
581 Configure compatibility hooks between Ezpl and classic logging/lib_mode.
583 Args:
584 hook_logger: If True, bridge stdlib logging records to Ezpl.
585 If False, remove previously installed stdlib bridges.
586 hook_printer: If True, ezpl.lib_mode.get_printer() delegates to
587 Ezpl's real printer once initialized. If False, it stays silent.
588 logger_names: Optional stdlib logger names to hook explicitly.
589 When omitted, root logger is targeted.
591 Notes:
592 - This method is safe to call multiple times (idempotent behavior).
593 - Hooking named loggers is useful for libraries with propagate=False.
594 """
595 type(self)._set_compatibility_hooks(
596 hook_logger=hook_logger,
597 hook_printer=hook_printer,
598 logger_names=logger_names,
599 )
601 # ------------------------------------------------
602 # CONFIG LOCK CONTROL
603 # ------------------------------------------------
605 @classmethod
606 def lock_config(cls) -> str:
607 """
608 Lock Ezpl configuration so that future configure() and set_level() calls
609 are blocked until unlock_config(token) is called.
611 Intended usage:
612 1. Root application configures Ezpl once
613 2. Calls token = Ezpl.lock_config()
614 3. Stores the token; libraries cannot reconfigure Ezpl without it
616 Returns:
617 str: A token required to unlock the configuration later.
618 """
619 with cls._lock:
620 if not cls._config_locked:
621 cls._config_locked = True
622 cls._config_lock_token = uuid4().hex
623 token = cls._config_lock_token
624 if token is None: 624 ↛ 626line 624 didn't jump to line 626 because the condition on line 624 was never true
625 # Should never happen: token is set whenever _config_locked is True
626 raise RuntimeError(
627 "Configuration lock token is None despite lock being active"
628 )
629 return token
631 @classmethod
632 def unlock_config(cls, token: str) -> bool:
633 """
634 Unlock Ezpl configuration.
636 Args:
637 token: The token returned by lock_config(). Must match exactly.
639 Returns:
640 True if unlocked successfully, False if the token is wrong.
641 """
642 with cls._lock:
643 if not cls._config_locked: 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true
644 return True
646 if token != cls._config_lock_token:
647 warnings.warn(
648 "Unlock denied: invalid token.",
649 UserWarning,
650 stacklevel=2,
651 )
652 return False
654 cls._config_locked = False
655 cls._config_lock_token = None
656 return True
658 @classmethod
659 def is_locked(cls) -> bool:
660 """Return True if the configuration is currently locked."""
661 return cls._config_locked
663 @classmethod
664 def is_lib_printer_hook_enabled(cls) -> bool:
665 """Return True when lib_mode printer proxy is allowed to delegate."""
666 return cls._lib_printer_hook_enabled
668 @classmethod
669 def _can_write_config(cls) -> bool:
670 """Return True when configuration writes are allowed."""
671 return not cls._config_locked
673 @classmethod
674 def _set_compatibility_hooks(
675 cls,
676 *,
677 hook_logger: bool,
678 hook_printer: bool,
679 logger_names: list[str] | None = None,
680 ) -> None:
681 """Apply logger/printer compatibility hooks with idempotent semantics."""
682 cls._lib_printer_hook_enabled = hook_printer
684 targets: list[str | None] = (
685 [None] if logger_names is None else list(dict.fromkeys(logger_names))
686 )
688 if hook_logger:
689 for logger_name in targets:
690 cls._install_intercept_handler(logger_name)
691 return
693 if logger_names is None: 693 ↛ 697line 693 didn't jump to line 697 because the condition on line 693 was always true
694 cls._remove_all_intercept_handlers()
695 return
697 for logger_name in targets:
698 cls._remove_intercept_handler(logger_name)
700 @classmethod
701 def _install_intercept_handler(cls, logger_name: str | None = None) -> None:
702 """
703 Install InterceptHandler on a stdlib logger.
705 Args:
706 logger_name: Name of the stdlib logger to hook. When None,
707 install on the root logger.
709 Notes:
710 This method is idempotent: calling it multiple times installs the
711 handler only once for a given target logger.
712 """
713 import logging as _logging
715 target = (
716 _logging.getLogger()
717 if logger_name is None
718 else _logging.getLogger(logger_name)
719 )
720 if not any(isinstance(h, InterceptHandler) for h in target.handlers):
721 target.addHandler(InterceptHandler())
722 target.setLevel(0)
724 # Ensure named loggers can still bubble to parent handlers if needed.
725 if logger_name is not None:
726 target.propagate = True
728 cls._stdlib_hook_targets.add(logger_name)
730 @classmethod
731 def _remove_intercept_handler(cls, logger_name: str | None = None) -> None:
732 """Remove InterceptHandler instances from a stdlib logger."""
733 import logging as _logging
735 target = (
736 _logging.getLogger()
737 if logger_name is None
738 else _logging.getLogger(logger_name)
739 )
740 target.handlers = [
741 handler
742 for handler in target.handlers
743 if not isinstance(handler, InterceptHandler)
744 ]
745 cls._stdlib_hook_targets.discard(logger_name)
747 @classmethod
748 def _remove_all_intercept_handlers(cls) -> None:
749 """Remove InterceptHandler from all tracked stdlib logger targets."""
750 tracked_targets = list(cls._stdlib_hook_targets)
751 if None not in tracked_targets:
752 tracked_targets.append(None)
754 for logger_name in tracked_targets:
755 cls._remove_intercept_handler(logger_name)
757 cls._stdlib_hook_targets.clear()
759 def set_log_file(self, log_file: Path | str) -> None:
760 """
761 Change the log file (requires reinitialization of the logger).
763 Args:
764 log_file: New path to the log file
766 Note: This will reinitialize the file logger but keep the singleton instance.
767 """
768 new_log_file = Path(log_file)
769 if new_log_file != self._log_file: 769 ↛ exitline 769 didn't return from function 'set_log_file' because the condition on line 769 was always true
770 self._log_file = new_log_file
771 # Update configuration
772 self._config_manager.set("log-file", str(new_log_file))
773 self._rebuild_logger()
775 def get_log_file(self) -> Path:
776 """
777 Get the current log file path.
779 Returns:
780 Path to the current log file
781 """
782 return self._log_file
784 def get_config(self) -> ConfigurationManager:
785 """
786 Get the current configuration manager.
788 Returns:
789 ConfigurationManager instance for accessing and modifying configuration
790 """
791 return self._config_manager
793 # ///////////////////////////////////////////////////////////////
794 # HANDLER OVERRIDE METHODS
795 # ///////////////////////////////////////////////////////////////
797 def set_printer_class(
798 self,
799 printer_class: type[EzPrinter] | EzPrinter,
800 **init_kwargs: Any,
801 ) -> None:
802 """
803 Replace the current printer with a custom printer class or instance.
805 Allows users to override the default printer with a custom class that
806 inherits from EzPrinter. The method preserves
807 current configuration values (level, indentation settings) unless
808 explicitly overridden in init_kwargs.
810 Args:
811 printer_class: Custom printer class inheriting from EzPrinter,
812 or an already instantiated EzPrinter instance
813 **init_kwargs: Optional initialization parameters for the printer
814 class. If not provided, current configuration values are used.
816 Raises:
817 TypeError: If printer_class is not a valid class or instance
818 ValidationError: If initialization parameters are invalid
820 Example:
821 >>> from ezpl import Ezpl, EzPrinter
822 >>>
823 >>> class CustomPrinter(EzPrinter):
824 ... def info(self, message):
825 ... super().info(f"[CUSTOM] {message}")
826 >>>
827 >>> ezpl = Ezpl()
828 >>> ezpl.set_printer_class(CustomPrinter, level="DEBUG")
829 >>> ezpl.get_printer().info("Test")
830 [CUSTOM] Test
831 """
832 from .core.exceptions import ValidationError
834 # If it's already an instance, use it directly
835 if isinstance(printer_class, EzPrinter):
836 new_printer = printer_class
837 # If it's a class, instantiate it
838 elif isinstance(printer_class, type):
839 # Validate that it's a subclass of EzPrinter
840 if not issubclass(printer_class, EzPrinter):
841 raise TypeError(
842 f"printer_class must be a subclass of {EzPrinter.__name__}, "
843 f"got {printer_class.__name__}"
844 )
846 # Preserve current configuration values if not provided
847 current_level = (
848 self._printer.level
849 if hasattr(self, "_printer") and self._printer
850 else self._config_manager.get_printer_level()
851 )
852 current_indent_step = (
853 self._printer.indent_step
854 if hasattr(self, "_printer") and self._printer
855 else self._config_manager.get_indent_step()
856 )
857 current_indent_symbol = (
858 self._printer.indent_symbol
859 if hasattr(self, "_printer") and self._printer
860 else self._config_manager.get_indent_symbol()
861 )
862 current_base_indent_symbol = (
863 self._printer.base_indent_symbol
864 if hasattr(self, "_printer") and self._printer
865 else self._config_manager.get_base_indent_symbol()
866 )
868 # Merge kwargs with default values
869 init_params = {
870 "level": init_kwargs.pop("level", current_level),
871 "indent_step": init_kwargs.pop("indent_step", current_indent_step),
872 "indent_symbol": init_kwargs.pop(
873 "indent_symbol", current_indent_symbol
874 ),
875 "base_indent_symbol": init_kwargs.pop(
876 "base_indent_symbol", current_base_indent_symbol
877 ),
878 }
879 init_params.update(init_kwargs)
881 # Create new instance
882 try:
883 new_printer = printer_class(**init_params)
884 except Exception as e:
885 raise ValidationError(
886 f"Failed to initialize printer class {printer_class.__name__}: {e}",
887 "printer_class",
888 str(printer_class),
889 ) from e
890 else:
891 raise TypeError(
892 f"printer_class must be a class or an instance of {EzPrinter.__name__}, "
893 f"got {type(printer_class).__name__}"
894 )
896 # Replace the instance
897 self._printer = new_printer
899 def set_logger_class(
900 self,
901 logger_class: type[EzLogger] | EzLogger,
902 **init_kwargs: Any,
903 ) -> None:
904 """
905 Replace the current logger with a custom logger class or instance.
907 Allows users to override the default logger with a custom class that
908 inherits from EzLogger. The method preserves current
909 configuration values (level, rotation, retention, compression) unless
910 explicitly overridden in init_kwargs.
912 Args:
913 logger_class: Custom logger class inheriting from EzLogger,
914 or an already instantiated EzLogger instance
915 **init_kwargs: Optional initialization parameters for the logger
916 class. If not provided, current configuration values are used.
918 Raises:
919 TypeError: If logger_class is not a valid class or instance
920 ValidationError: If initialization parameters are invalid
921 FileOperationError: If file operations fail during logger creation
922 (may be raised by the logger class constructor)
924 Example:
925 >>> from ezpl import Ezpl, EzLogger
926 >>>
927 >>> class CustomLogger(EzLogger):
928 ... def info(self, message):
929 ... super().info(f"[CUSTOM LOG] {message}")
930 >>>
931 >>> ezpl = Ezpl()
932 >>> ezpl.set_logger_class(CustomLogger, log_file="custom.log")
933 >>> ezpl.get_logger().info("Test")
934 """
935 from .core.exceptions import ValidationError
937 # If it's already an instance, use it directly
938 if isinstance(logger_class, EzLogger):
939 new_logger = logger_class
940 # If it's a class, instantiate it
941 elif isinstance(logger_class, type):
942 # Validate that it's a subclass of EzLogger
943 if not issubclass(logger_class, EzLogger):
944 raise TypeError(
945 f"logger_class must be a subclass of {EzLogger.__name__}, "
946 f"got {logger_class.__name__}"
947 )
949 # Preserve current configuration values if not provided
950 current_level = (
951 self._logger.level
952 if hasattr(self, "_logger") and self._logger
953 else self._config_manager.get_file_logger_level()
954 )
955 current_log_file = (
956 self._log_file
957 if hasattr(self, "_log_file")
958 else self._config_manager.get_log_file()
959 )
960 current_rotation = (
961 self._logger.rotation
962 if hasattr(self, "_logger") and self._logger
963 else self._config_manager.get_log_rotation()
964 )
965 current_retention = (
966 self._logger.retention
967 if hasattr(self, "_logger") and self._logger
968 else self._config_manager.get_log_retention()
969 )
970 current_compression = (
971 self._logger.compression
972 if hasattr(self, "_logger") and self._logger
973 else self._config_manager.get_log_compression()
974 )
976 # Merge kwargs with default values
977 init_params = {
978 "log_file": init_kwargs.pop("log_file", current_log_file),
979 "level": init_kwargs.pop("level", current_level),
980 "rotation": init_kwargs.pop("rotation", current_rotation),
981 "retention": init_kwargs.pop("retention", current_retention),
982 "compression": init_kwargs.pop("compression", current_compression),
983 }
984 init_params.update(init_kwargs)
986 # Close previous logger before creating new one to avoid resource leaks
987 try:
988 if hasattr(self, "_logger") and self._logger:
989 self._logger.close()
990 except (EzplError, OSError, RuntimeError) as e:
991 logger.error(f"Error while closing previous logger: {e}")
993 # Create new instance
994 try:
995 new_logger = logger_class(**init_params)
996 except Exception as e:
997 raise ValidationError(
998 f"Failed to initialize logger class {logger_class.__name__}: {e}",
999 "logger_class",
1000 str(logger_class),
1001 ) from e
1002 else:
1003 raise TypeError(
1004 f"logger_class must be a class or an instance of {EzLogger.__name__}, "
1005 f"got {type(logger_class).__name__}"
1006 )
1008 # Replace the instance
1009 self._logger = new_logger
1011 # ///////////////////////////////////////////////////////////////
1012 # CONFIGURATION METHODS
1013 # ///////////////////////////////////////////////////////////////
1015 def reload_config(self) -> None:
1016 """
1017 Reload configuration from file and environment variables.
1019 This method reloads the configuration and reapplies it to handlers.
1020 Useful when environment variables or the config file have changed
1021 after the singleton was initialized.
1023 Note: This will reinitialize handlers with the new configuration.
1024 """
1025 # Reload configuration
1026 self._config_manager.reload()
1028 # Get configuration values
1029 printer_level = self._config_manager.get_printer_level()
1030 file_logger_level = self._config_manager.get_file_logger_level()
1031 global_log_level = self._config_manager.get_log_level()
1033 # Check if specific levels are explicitly set (not just defaults)
1034 # Priority: specific levels > global level
1035 # Only apply global level if specific levels are not explicitly set
1036 printer_level_explicit = self._config_manager.has_key("printer-level")
1037 file_logger_level_explicit = self._config_manager.has_key("file-logger-level")
1038 global_log_level_explicit = self._config_manager.has_key("log-level")
1040 # Respect manually set levels: don't override if set via set_level()
1041 printer_manually_set = self._printer.level_manually_set
1042 logger_manually_set = self._logger.level_manually_set
1044 # Reapply to handlers with priority logic (skip if manually set)
1045 if not printer_manually_set: 1045 ↛ 1046line 1045 didn't jump to line 1046 because the condition on line 1045 was never true
1046 effective_printer = (
1047 printer_level
1048 if printer_level_explicit
1049 else global_log_level
1050 if global_log_level_explicit
1051 else printer_level
1052 )
1053 self.set_printer_level(effective_printer)
1055 if not logger_manually_set: 1055 ↛ 1056line 1055 didn't jump to line 1056 because the condition on line 1055 was never true
1056 effective_logger = (
1057 file_logger_level
1058 if file_logger_level_explicit
1059 else (
1060 global_log_level if global_log_level_explicit else file_logger_level
1061 )
1062 )
1063 self.set_logger_level(effective_logger)
1065 # Reinitialize handlers — new instances have _level_manually_set = False by default
1066 self._rebuild_logger()
1067 self._rebuild_printer()
1069 def configure(
1070 self,
1071 config_dict: dict[str, Any] | None = None,
1072 *,
1073 persist: bool = False,
1074 **kwargs: Any,
1075 ) -> bool:
1076 """
1077 Configure Ezpl dynamically.
1079 Args:
1080 config_dict: Dictionary of configuration values to update
1081 persist: If True, write changes to ~/.ezpl/config.json so they
1082 survive future runs. Defaults to False (in-memory only).
1083 **kwargs: Configuration options (alternative to config_dict):
1084 - log_file or log-file: Path to log file
1085 - printer_level or printer-level: Printer log level
1086 - file_logger_level or file-logger-level: File logger level
1087 - level or log-level: Set both printer and logger level
1088 - log_rotation or log-rotation: Rotation setting (e.g., "10 MB", "1 day")
1089 - log_retention or log-retention: Retention period (e.g., "7 days")
1090 - log_compression or log-compression: Compression format (e.g., "zip", "gz")
1091 - indent_step or indent-step: Indentation step size
1092 - indent_symbol or indent-symbol: Symbol for indentation
1093 - base_indent_symbol or base-indent-symbol: Base indentation symbol
1095 Returns:
1096 True if configuration was applied, False if it was blocked by lock.
1097 """
1098 # Merge config_dict and kwargs
1099 if config_dict:
1100 kwargs.update(config_dict)
1102 # If configuration is locked, warn and return False
1103 if not self._can_write_config():
1104 warnings.warn(
1105 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) "
1106 "to unlock before reconfiguring.",
1107 UserWarning,
1108 stacklevel=2,
1109 )
1110 return False
1112 # Normalize keys: convert underscores to hyphens for consistency
1113 normalized_config = {}
1114 key_mapping = {
1115 "log_file": "log-file",
1116 "printer_level": "printer-level",
1117 "file_logger_level": "file-logger-level",
1118 "level": "log-level",
1119 "log_rotation": "log-rotation",
1120 "log_retention": "log-retention",
1121 "log_compression": "log-compression",
1122 "indent_step": "indent-step",
1123 "indent_symbol": "indent-symbol",
1124 "base_indent_symbol": "base-indent-symbol",
1125 }
1127 for key, value in kwargs.items():
1128 # Use normalized key if mapping exists, otherwise keep original
1129 normalized_key = key_mapping.get(key, key)
1130 normalized_config[normalized_key] = value
1132 # Extract log-file before update — set_log_file() owns that write.
1133 # Keep it persisted in config when requested.
1134 new_log_file = normalized_config.pop("log-file", None)
1135 if new_log_file is not None:
1136 normalized_config["log-file"] = str(Path(new_log_file))
1138 # Update configuration manager
1139 self._config_manager.update(normalized_config)
1140 # Apply changes to handlers
1141 if new_log_file is not None:
1142 self.set_log_file(new_log_file)
1144 # Handle log level changes with priority: specific > global
1145 self._apply_level_priority(
1146 printer_level=normalized_config.get("printer-level"),
1147 file_logger_level=normalized_config.get("file-logger-level"),
1148 global_level=normalized_config.get("log-level"),
1149 )
1151 # Reinitialize logger if rotation settings changed
1152 rotation_changed = any(
1153 key in normalized_config
1154 for key in ["log-rotation", "log-retention", "log-compression"]
1155 )
1156 if rotation_changed:
1157 self._rebuild_logger()
1159 # Reinitialize printer if indent settings changed
1160 indent_changed = any(
1161 key in normalized_config
1162 for key in ["indent-step", "indent-symbol", "base-indent-symbol"]
1163 )
1164 if indent_changed: 1164 ↛ 1165line 1164 didn't jump to line 1165 because the condition on line 1164 was never true
1165 self._rebuild_printer()
1167 if persist:
1168 self._config_manager.save()
1170 return True
1173# ///////////////////////////////////////////////////////////////
1174# PUBLIC API
1175# ///////////////////////////////////////////////////////////////
1177__all__ = [
1178 "Ezpl",
1179]