Coverage for src/ezplog/ezpl.py: 72.00%
327 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-16 19:04 +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_CONFIG_LOCKED_WARNING = (
37 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock."
38)
40# ///////////////////////////////////////////////////////////////
41# CLASSES
42# ///////////////////////////////////////////////////////////////
45class Ezpl:
46 """
47 Main logging singleton for the Ezpl framework.
49 Ezpl provides a unified, thread-safe interface for console and file logging
50 with advanced features including indentation management, pattern-based logging,
51 and dynamic progress bars. It implements the Singleton pattern to ensure only
52 one instance exists throughout the application lifecycle.
54 Attributes:
55 _instance: The singleton instance of Ezpl
56 _lock: Thread lock for synchronized access
57 _config_locked: Whether configuration can be modified
58 _log_file: Path to the log file
59 _printer: Console output handler
60 _logger: File logging handler
61 _config_manager: Configuration manager instance
63 Note:
64 Once initialized, Ezpl cannot be re-configured unless reset.
65 Access it via the singleton pattern or module-level functions.
67 Example:
68 >>> log = Ezpl()
69 >>> log.printer.log("INFO", "Application started")
70 >>> log.logger.log("INFO", "Starting logging to file")
71 """
73 _instance: Ezpl | None = None
74 _lock: threading.RLock = threading.RLock()
75 _config_locked: bool = False
76 _config_lock_token: str | None = None
77 _log_file: Path
78 _printer: EzPrinter
79 _logger: EzLogger
80 _config_manager: ConfigurationManager
81 _lib_printer_hook_enabled: bool = True
82 _stdlib_hook_targets: set[str | None] = set()
84 # ///////////////////////////////////////////////////////////////
85 # INIT
86 # ///////////////////////////////////////////////////////////////
88 @classmethod
89 def is_initialized(cls) -> bool:
90 """
91 Return True if the Ezpl singleton has already been created.
93 Useful for libraries that want to know whether they are the first
94 to initialize Ezpl or if they should avoid re-configuring it.
95 """
96 return cls._instance is not None
98 def __new__(
99 cls,
100 log_file: Path | str | None = None,
101 log_level: str | None = None,
102 printer_level: str | None = None,
103 file_logger_level: str | None = None,
104 log_rotation: str | None = None,
105 log_retention: str | None = None,
106 log_compression: str | None = None,
107 log_backtrace: bool | None = None,
108 log_diagnose: bool | None = None,
109 indent_step: int | None = None,
110 indent_symbol: str | None = None,
111 base_indent_symbol: str | None = None,
112 *,
113 lock_config: bool = False,
114 hook_logger: bool = True,
115 hook_printer: bool = True,
116 logger_names: list[str] | None = None,
117 ) -> Ezpl:
118 """
119 Creates and returns a new instance of Ezpl if none exists.
121 **Notes:**
122 Ensures only one instance of Ezpl exists (Singleton pattern).
124 **Priority order for configuration (for each parameter):**
125 1. Arguments passed directly (highest priority)
126 2. Environment variables (EZPL_*)
127 3. Configuration file (~/.ezpl/config.json)
128 4. Default values (lowest priority)
130 **Args:**
132 * `log_file` (Path | str, optional): Path to the log file
133 * `log_level` (str, optional): Global log level (applies to both printer and logger)
134 * `printer_level` (str, optional): Printer log level
135 * `file_logger_level` (str, optional): File logger level
136 * `log_rotation` (str, optional): Rotation setting (e.g., "10 MB", "1 day")
137 * `log_retention` (str, optional): Retention period (e.g., "7 days")
138 * `log_compression` (str, optional): Compression format (e.g., "zip", "gz")
139 * `log_backtrace` (bool, optional): Extend tracebacks beyond the catching point
140 * `log_diagnose` (bool, optional): Include local variable values in tracebacks
141 * `indent_step` (int, optional): Indentation step size
142 * `indent_symbol` (str, optional): Symbol for indentation
143 * `base_indent_symbol` (str, optional): Base indentation symbol
144 * `lock_config` (bool): If True, lock configuration immediately after init.
145 Subsequent configure() calls will be blocked unless a valid token is used.
146 The lock token is stored in Ezpl._config_lock_token.
147 * `hook_logger` (bool): If True, install stdlib logging bridge(s) so
148 classic loggers are forwarded to the loguru pipeline.
149 * `hook_printer` (bool): If True, allow ezpl.lib_mode.get_printer()
150 to delegate to the real EzPrinter instance once initialized.
151 * `logger_names` (list[str] | None): Optional stdlib logger names to
152 bridge explicitly. When omitted, the root logger is used.
154 **Returns:**
156 * `Ezpl`: The singleton instance of the Ezpl class.
158 **Raises:**
160 * `None`.
161 """
163 # //////
164 # Double-checked locking pattern for thread-safe singleton
165 if cls._instance is None:
166 with cls._lock:
167 # Check again after acquiring lock (double-checked locking)
168 if cls._instance is None:
169 # Initialize configuration manager
170 cls._config_manager = ConfigurationManager()
172 # Determine configuration values with priority: arg > env > config file > default
173 # Helper function to get value with priority order
174 def get_config_value(
175 arg_value, config_key: str, getter_method
176 ) -> Any:
177 """
178 Get configuration value with priority: arg > env > config file > default
180 Args:
181 arg_value: Value from argument (can be None)
182 config_key: Configuration key name
183 getter_method: Method to get default value from config manager
185 Returns:
186 Final configuration value
187 """
188 # Priority 1: Argument direct
189 if arg_value is not None:
190 return arg_value
192 # Priority 2: Environment variable (already loaded in config_manager)
193 # Priority 3: Config file (already loaded in config_manager)
194 # Priority 4: Default (via getter method)
195 # The config_manager already has the correct priority (env > file > default)
196 config_value = cls._config_manager.get(config_key)
197 if config_value is not None: 197 ↛ 201line 197 didn't jump to line 201 because the condition on line 197 was always true
198 return config_value
200 # Fallback to default via getter
201 return getter_method()
203 # Log file
204 if log_file:
205 cls._log_file = Path(log_file)
206 else:
207 cls._log_file = cls._config_manager.get_log_file()
209 # Log level (global)
210 final_log_level = get_config_value(
211 log_level, "log-level", cls._config_manager.get_log_level
212 )
214 # Printer level
215 final_printer_level = get_config_value(
216 printer_level,
217 "printer-level",
218 cls._config_manager.get_printer_level,
219 )
221 # File logger level
222 final_file_logger_level = get_config_value(
223 file_logger_level,
224 "file-logger-level",
225 cls._config_manager.get_file_logger_level,
226 )
228 # Rotation settings (can be None)
229 # Priority: arg > env > config file > default
230 # Note: If arg is None (default), we check env/config/default
231 # If user wants to explicitly set None, they can pass None or use configure()
232 final_rotation = (
233 log_rotation
234 if log_rotation is not None
235 else cls._config_manager.get_log_rotation()
236 )
237 final_retention = (
238 log_retention
239 if log_retention is not None
240 else cls._config_manager.get_log_retention()
241 )
242 final_compression = (
243 log_compression
244 if log_compression is not None
245 else cls._config_manager.get_log_compression()
246 )
247 final_backtrace = (
248 log_backtrace
249 if log_backtrace is not None
250 else cls._config_manager.get_log_backtrace()
251 )
252 final_diagnose = (
253 log_diagnose
254 if log_diagnose is not None
255 else cls._config_manager.get_log_diagnose()
256 )
258 # Indent settings
259 final_indent_step = get_config_value(
260 indent_step, "indent-step", cls._config_manager.get_indent_step
261 )
262 final_indent_symbol = get_config_value(
263 indent_symbol,
264 "indent-symbol",
265 cls._config_manager.get_indent_symbol,
266 )
267 final_base_indent_symbol = get_config_value(
268 base_indent_symbol,
269 "base-indent-symbol",
270 cls._config_manager.get_base_indent_symbol,
271 )
273 instance = object.__new__(cls)
274 cls._instance = instance
276 # Initialize printer with resolved configuration
277 cls._printer = EzPrinter(
278 level=final_printer_level,
279 indent_step=final_indent_step,
280 indent_symbol=final_indent_symbol,
281 base_indent_symbol=final_base_indent_symbol,
282 )
284 # Initialize logger with resolved configuration
285 cls._logger = EzLogger(
286 log_file=cls._log_file,
287 level=final_file_logger_level,
288 rotation=final_rotation,
289 retention=final_retention,
290 compression=final_compression,
291 backtrace=final_backtrace,
292 diagnose=final_diagnose,
293 )
295 # Apply global log level with priority: specific > global
296 cls._instance._apply_level_priority(
297 printer_level=printer_level,
298 file_logger_level=file_logger_level,
299 global_level=final_log_level,
300 )
302 # Apply post-init options — inside the critical section,
303 # first construction only (double-checked locking guarantees this)
304 if lock_config: 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true
305 cls._config_lock_token = cls.lock_config()
307 cls._set_compatibility_hooks(
308 hook_logger=hook_logger,
309 hook_printer=hook_printer,
310 logger_names=logger_names,
311 )
313 # Type narrowing: _instance is guaranteed to be set at this point
314 if cls._instance is None: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true
315 raise InitializationError("Singleton initialization failed unexpectedly")
316 return cls._instance
318 # ///////////////////////////////////////////////////////////////
319 # PRIVATE HELPERS
320 # ///////////////////////////////////////////////////////////////
322 def _apply_level_priority(
323 self,
324 *,
325 printer_level: str | None = None,
326 file_logger_level: str | None = None,
327 global_level: str | None = None,
328 ) -> None:
329 """
330 Apply log levels with priority: specific level > global level.
332 Only sets levels when a non-None value can be resolved.
333 If a specific level is not provided, global_level is used as fallback.
334 """
335 effective_printer = printer_level if printer_level is not None else global_level
336 effective_logger = (
337 file_logger_level if file_logger_level is not None else global_level
338 )
340 if effective_printer:
341 self.set_printer_level(effective_printer)
343 if effective_logger:
344 self.set_logger_level(effective_logger)
346 def _rebuild_logger(
347 self,
348 *,
349 level: str | None = None,
350 ) -> None:
351 """
352 Close the current file logger and reinitialize it with current configuration.
354 Args:
355 level: Override the log level. If None, preserves the current level.
356 """
357 current_level = (
358 level
359 or (
360 self._logger.level
361 if hasattr(self, "_logger") and self._logger
362 else None
363 )
364 or self._config_manager.get_file_logger_level()
365 )
366 try:
367 if hasattr(self, "_logger") and self._logger: 367 ↛ 371line 367 didn't jump to line 371 because the condition on line 367 was always true
368 self._logger.close()
369 except (EzplError, OSError, RuntimeError) as e:
370 logger.error(f"Error while closing logger: {e}")
371 self._logger = EzLogger(
372 log_file=self._log_file,
373 level=current_level,
374 rotation=self._config_manager.get_log_rotation(),
375 retention=self._config_manager.get_log_retention(),
376 compression=self._config_manager.get_log_compression(),
377 backtrace=self._config_manager.get_log_backtrace(),
378 diagnose=self._config_manager.get_log_diagnose(),
379 )
381 def _rebuild_printer(
382 self,
383 *,
384 level: str | None = None,
385 ) -> None:
386 """
387 Reinitialize the console printer with current configuration.
389 Args:
390 level: Override the log level. If None, preserves the current level.
391 """
392 current_level = (
393 level
394 or (
395 self._printer.level
396 if hasattr(self, "_printer") and self._printer
397 else None
398 )
399 or self._config_manager.get_printer_level()
400 )
401 self._printer = EzPrinter(
402 level=current_level,
403 indent_step=self._config_manager.get_indent_step(),
404 indent_symbol=self._config_manager.get_indent_symbol(),
405 base_indent_symbol=self._config_manager.get_base_indent_symbol(),
406 )
408 # ///////////////////////////////////////////////////////////////
409 # GETTER
410 # ///////////////////////////////////////////////////////////////
412 def get_printer(self) -> EzPrinter:
413 """
414 Returns the EzPrinter instance.
416 **Returns:**
418 * EzPrinter: The console printer instance providing info(), debug(), success(), etc.
419 Implements PrinterProtocol for type safety.
420 """
421 return self._printer
423 # ///////////////////////////////////////////////////////////////
425 def get_logger(self) -> EzLogger:
426 """
427 Returns the EzLogger instance.
429 **Returns:**
431 * EzLogger: The file logger instance for file logging.
432 Use logger.info(), logger.debug(), etc. directly.
433 For advanced loguru features, use logger.get_loguru()
434 Implements LoggerProtocol for type safety.
435 """
436 return self._logger
438 # ///////////////////////////////////////////////////////////////
439 # UTILS METHODS
440 # ///////////////////////////////////////////////////////////////
442 @property
443 def printer_level(self) -> str:
444 """Return the current printer logging level."""
445 return self._printer.level
447 @property
448 def logger_level(self) -> str:
449 """Return the current file logger logging level."""
450 return self._logger.level
452 def set_level(self, level: str) -> None:
453 """
454 Set the log level for both the printer and the logger simultaneously.
456 **Args:**
458 * `level` (str): The desired log level (e.g., "INFO", "WARNING").
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 _CONFIG_LOCKED_WARNING,
467 UserWarning,
468 stacklevel=2,
469 )
470 return
471 self.set_logger_level(level)
472 self.set_printer_level(level)
474 def set_printer_level(self, level: str) -> None:
475 """
476 Set the log level for the printer only.
478 **Args:**
480 * `level` (str): The desired log level for the printer.
482 **Returns:**
484 * `None`.
485 """
486 if not self._can_write_config(): 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true
487 warnings.warn(
488 _CONFIG_LOCKED_WARNING,
489 UserWarning,
490 stacklevel=2,
491 )
492 return
493 self._printer.set_level(level)
495 def set_logger_level(self, level: str) -> None:
496 """
497 Set the log level for the file logger only.
499 **Args:**
501 * `level` (str): The desired log level for the file logger.
503 **Returns:**
505 * `None`.
506 """
507 if not self._can_write_config(): 507 ↛ 508line 507 didn't jump to line 508 because the condition on line 507 was never true
508 warnings.warn(
509 _CONFIG_LOCKED_WARNING,
510 UserWarning,
511 stacklevel=2,
512 )
513 return
514 self._logger.set_level(level)
516 # ///////////////////////////////////////////////////////////////
517 # FACADE METHODS
518 # ///////////////////////////////////////////////////////////////
520 def trace(self, message: Any, *args: Any, **kwargs: Any) -> None:
521 """Log a trace message to the console printer."""
522 self._printer.trace(message, *args, **kwargs)
524 def debug(self, message: Any, *args: Any, **kwargs: Any) -> None:
525 """Log a debug message to the console printer."""
526 self._printer.debug(message, *args, **kwargs)
528 def info(self, message: Any, *args: Any, **kwargs: Any) -> None:
529 """Log an info message to the console printer."""
530 self._printer.info(message, *args, **kwargs)
532 def success(self, message: Any, *args: Any, **kwargs: Any) -> None:
533 """Log a success message to the console printer."""
534 self._printer.success(message, *args, **kwargs)
536 def warning(self, message: Any, *args: Any, **kwargs: Any) -> None:
537 """Log a warning message to the console printer."""
538 self._printer.warning(message, *args, **kwargs)
540 def error(
541 self, message: Any, *args: Any, exc_info: bool = False, **kwargs: Any
542 ) -> None:
543 """Log an error message to the console printer.
545 Args:
546 message: Message to display.
547 *args: Positional arguments for `{}` formatting.
548 exc_info: If True, also render the active exception traceback.
549 **kwargs: Keyword arguments for `{}` formatting.
550 """
551 self._printer.error(message, *args, exc_info=exc_info, **kwargs)
553 def critical(
554 self, message: Any, *args: Any, exc_info: bool = False, **kwargs: Any
555 ) -> None:
556 """Log a critical message to the console printer.
558 Args:
559 message: Message to display.
560 *args: Positional arguments for `{}` formatting.
561 exc_info: If True, also render the active exception traceback.
562 **kwargs: Keyword arguments for `{}` formatting.
563 """
564 self._printer.critical(message, *args, exc_info=exc_info, **kwargs)
566 def exception(self, message: Any, *args: Any, **kwargs: Any) -> None:
567 """Log an error message with the active exception traceback (console)."""
568 self._printer.exception(message, *args, **kwargs)
570 def log(self, level: str, message: Any, *args: Any, **kwargs: Any) -> None:
571 """Log a message at the given level to the console printer.
573 Args:
574 level: Log level name (TRACE, DEBUG, INFO, SUCCESS, WARNING,
575 ERROR, CRITICAL).
576 message: Message to display.
577 *args: Positional arguments for `{}` formatting.
578 **kwargs: Keyword arguments for `{}` formatting.
579 """
580 self._printer.log(level, message, *args, **kwargs)
582 # ///////////////////////////////////////////////////////////////
584 def add_separator(self) -> None:
585 """
586 Adds a separator to the log file.
588 **Returns:**
590 * `None`.
591 """
592 self._logger.add_separator()
594 # ///////////////////////////////////////////////////////////////
596 @contextmanager
597 def manage_indent(self) -> Generator[None]:
598 """
599 Context manager to manage indentation level.
601 **Returns:**
603 * `None`.
604 """
605 with self._printer.manage_indent():
606 yield
608 # ///////////////////////////////////////////////////////////////
609 # ENHANCED METHODS
610 # ///////////////////////////////////////////////////////////////
612 @classmethod
613 def reset(cls) -> None:
614 """
615 Reset the singleton instance (useful for testing).
617 Warning: This will destroy the current instance and all its state.
618 """
619 if cls._instance is not None:
620 # Close logger handlers to release file handles (important on Windows)
621 try:
622 if hasattr(cls._instance, "_logger") and cls._instance._logger: 622 ↛ 626line 622 didn't jump to line 626 because the condition on line 622 was always true
623 cls._instance._logger.close()
624 except (EzplError, OSError, RuntimeError) as e:
625 logger.error(f"Error during cleanup: {e}")
626 cls._instance = None
627 cls._remove_all_intercept_handlers()
628 cls._lib_printer_hook_enabled = True
629 # Also reset configuration lock
630 cls._config_locked = False
631 cls._config_lock_token = None
633 def set_compatibility_hooks(
634 self,
635 *,
636 hook_logger: bool = True,
637 hook_printer: bool = True,
638 logger_names: list[str] | None = None,
639 ) -> None:
640 """
641 Configure compatibility hooks between Ezpl and classic logging/lib_mode.
643 Args:
644 hook_logger: If True, bridge stdlib logging records to Ezpl.
645 If False, remove previously installed stdlib bridges.
646 hook_printer: If True, ezpl.lib_mode.get_printer() delegates to
647 Ezpl's real printer once initialized. If False, it stays silent.
648 logger_names: Optional stdlib logger names to hook explicitly.
649 When omitted, root logger is targeted.
651 Notes:
652 - This method is safe to call multiple times (idempotent behavior).
653 - Hooking named loggers is useful for libraries with propagate=False.
654 """
655 type(self)._set_compatibility_hooks(
656 hook_logger=hook_logger,
657 hook_printer=hook_printer,
658 logger_names=logger_names,
659 )
661 # ------------------------------------------------
662 # CONFIG LOCK CONTROL
663 # ------------------------------------------------
665 @classmethod
666 def lock_config(cls) -> str:
667 """
668 Lock Ezpl configuration so that future configure() and set_level() calls
669 are blocked until unlock_config(token) is called.
671 Intended usage:
672 1. Root application configures Ezpl once
673 2. Calls token = Ezpl.lock_config()
674 3. Stores the token; libraries cannot reconfigure Ezpl without it
676 Returns:
677 str: A token required to unlock the configuration later.
678 """
679 with cls._lock:
680 if not cls._config_locked:
681 cls._config_locked = True
682 cls._config_lock_token = uuid4().hex
683 token = cls._config_lock_token
684 if token is None: 684 ↛ 686line 684 didn't jump to line 686 because the condition on line 684 was never true
685 # Should never happen: token is set whenever _config_locked is True
686 raise RuntimeError(
687 "Configuration lock token is None despite lock being active"
688 )
689 return token
691 @classmethod
692 def unlock_config(cls, token: str) -> bool:
693 """
694 Unlock Ezpl configuration.
696 Args:
697 token: The token returned by lock_config(). Must match exactly.
699 Returns:
700 True if unlocked successfully, False if the token is wrong.
701 """
702 with cls._lock:
703 if not cls._config_locked: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true
704 return True
706 if token != cls._config_lock_token:
707 warnings.warn(
708 "Unlock denied: invalid token.",
709 UserWarning,
710 stacklevel=2,
711 )
712 return False
714 cls._config_locked = False
715 cls._config_lock_token = None
716 return True
718 @classmethod
719 def is_locked(cls) -> bool:
720 """Return True if the configuration is currently locked."""
721 return cls._config_locked
723 @classmethod
724 def is_lib_printer_hook_enabled(cls) -> bool:
725 """Return True when lib_mode printer proxy is allowed to delegate."""
726 return cls._lib_printer_hook_enabled
728 @classmethod
729 def _can_write_config(cls) -> bool:
730 """Return True when configuration writes are allowed."""
731 return not cls._config_locked
733 @classmethod
734 def _set_compatibility_hooks(
735 cls,
736 *,
737 hook_logger: bool,
738 hook_printer: bool,
739 logger_names: list[str] | None = None,
740 ) -> None:
741 """Apply logger/printer compatibility hooks with idempotent semantics."""
742 cls._lib_printer_hook_enabled = hook_printer
744 targets: list[str | None] = (
745 [None] if logger_names is None else list(dict.fromkeys(logger_names))
746 )
748 if hook_logger:
749 for logger_name in targets:
750 cls._install_intercept_handler(logger_name)
751 return
753 if logger_names is None: 753 ↛ 757line 753 didn't jump to line 757 because the condition on line 753 was always true
754 cls._remove_all_intercept_handlers()
755 return
757 for logger_name in targets:
758 cls._remove_intercept_handler(logger_name)
760 @classmethod
761 def _install_intercept_handler(cls, logger_name: str | None = None) -> None:
762 """
763 Install InterceptHandler on a stdlib logger.
765 Args:
766 logger_name: Name of the stdlib logger to hook. When None,
767 install on the root logger.
769 Notes:
770 This method is idempotent: calling it multiple times installs the
771 handler only once for a given target logger.
772 """
773 import logging as _logging
775 target = (
776 _logging.getLogger()
777 if logger_name is None
778 else _logging.getLogger(logger_name)
779 )
780 if not any(isinstance(h, InterceptHandler) for h in target.handlers):
781 target.addHandler(InterceptHandler())
782 target.setLevel(0)
784 # Ensure named loggers can still bubble to parent handlers if needed.
785 if logger_name is not None:
786 target.propagate = True
788 cls._stdlib_hook_targets.add(logger_name)
790 @classmethod
791 def _remove_intercept_handler(cls, logger_name: str | None = None) -> None:
792 """Remove InterceptHandler instances from a stdlib logger."""
793 import logging as _logging
795 target = (
796 _logging.getLogger()
797 if logger_name is None
798 else _logging.getLogger(logger_name)
799 )
800 target.handlers = [
801 handler
802 for handler in target.handlers
803 if not isinstance(handler, InterceptHandler)
804 ]
805 cls._stdlib_hook_targets.discard(logger_name)
807 @classmethod
808 def _remove_all_intercept_handlers(cls) -> None:
809 """Remove InterceptHandler from all tracked stdlib logger targets."""
810 tracked_targets = list(cls._stdlib_hook_targets)
811 if None not in tracked_targets:
812 tracked_targets.append(None)
814 for logger_name in tracked_targets:
815 cls._remove_intercept_handler(logger_name)
817 cls._stdlib_hook_targets.clear()
819 def set_log_file(self, log_file: Path | str) -> None:
820 """
821 Change the log file (requires reinitialization of the logger).
823 Args:
824 log_file: New path to the log file
826 Note: This will reinitialize the file logger but keep the singleton instance.
827 """
828 new_log_file = Path(log_file)
829 if new_log_file != self._log_file: 829 ↛ exitline 829 didn't return from function 'set_log_file' because the condition on line 829 was always true
830 self._log_file = new_log_file
831 # Update configuration
832 self._config_manager.set("log-file", str(new_log_file))
833 self._rebuild_logger()
835 def get_log_file(self) -> Path:
836 """
837 Get the current log file path.
839 Returns:
840 Path to the current log file
841 """
842 return self._log_file
844 def get_config(self) -> ConfigurationManager:
845 """
846 Get the current configuration manager.
848 Returns:
849 ConfigurationManager instance for accessing and modifying configuration
850 """
851 return self._config_manager
853 # ///////////////////////////////////////////////////////////////
854 # HANDLER OVERRIDE METHODS
855 # ///////////////////////////////////////////////////////////////
857 def set_printer_class(
858 self,
859 printer_class: type[EzPrinter] | EzPrinter,
860 **init_kwargs: Any,
861 ) -> None:
862 """
863 Replace the current printer with a custom printer class or instance.
865 Allows users to override the default printer with a custom class that
866 inherits from EzPrinter. The method preserves
867 current configuration values (level, indentation settings) unless
868 explicitly overridden in init_kwargs.
870 Args:
871 printer_class: Custom printer class inheriting from EzPrinter,
872 or an already instantiated EzPrinter instance
873 **init_kwargs: Optional initialization parameters for the printer
874 class. If not provided, current configuration values are used.
876 Raises:
877 TypeError: If printer_class is not a valid class or instance
878 ValidationError: If initialization parameters are invalid
880 Example:
881 >>> from ezpl import Ezpl, EzPrinter
882 >>>
883 >>> class CustomPrinter(EzPrinter):
884 ... def info(self, message):
885 ... super().info(f"[CUSTOM] {message}")
886 >>>
887 >>> ezpl = Ezpl()
888 >>> ezpl.set_printer_class(CustomPrinter, level="DEBUG")
889 >>> ezpl.get_printer().info("Test")
890 [CUSTOM] Test
891 """
892 from .core.exceptions import ValidationError
894 # If it's already an instance, use it directly
895 if isinstance(printer_class, EzPrinter):
896 new_printer = printer_class
897 # If it's a class, instantiate it
898 elif isinstance(printer_class, type):
899 # Validate that it's a subclass of EzPrinter
900 if not issubclass(printer_class, EzPrinter):
901 raise TypeError(
902 f"printer_class must be a subclass of {EzPrinter.__name__}, "
903 f"got {printer_class.__name__}"
904 )
906 # Preserve current configuration values if not provided
907 current_level = (
908 self._printer.level
909 if hasattr(self, "_printer") and self._printer
910 else self._config_manager.get_printer_level()
911 )
912 current_indent_step = (
913 self._printer.indent_step
914 if hasattr(self, "_printer") and self._printer
915 else self._config_manager.get_indent_step()
916 )
917 current_indent_symbol = (
918 self._printer.indent_symbol
919 if hasattr(self, "_printer") and self._printer
920 else self._config_manager.get_indent_symbol()
921 )
922 current_base_indent_symbol = (
923 self._printer.base_indent_symbol
924 if hasattr(self, "_printer") and self._printer
925 else self._config_manager.get_base_indent_symbol()
926 )
928 # Merge kwargs with default values
929 init_params = {
930 "level": init_kwargs.pop("level", current_level),
931 "indent_step": init_kwargs.pop("indent_step", current_indent_step),
932 "indent_symbol": init_kwargs.pop(
933 "indent_symbol", current_indent_symbol
934 ),
935 "base_indent_symbol": init_kwargs.pop(
936 "base_indent_symbol", current_base_indent_symbol
937 ),
938 }
939 init_params.update(init_kwargs)
941 # Create new instance
942 try:
943 new_printer = printer_class(**init_params)
944 except Exception as e:
945 raise ValidationError(
946 f"Failed to initialize printer class {printer_class.__name__}: {e}",
947 "printer_class",
948 str(printer_class),
949 ) from e
950 else:
951 raise TypeError(
952 f"printer_class must be a class or an instance of {EzPrinter.__name__}, "
953 f"got {type(printer_class).__name__}"
954 )
956 # Replace the instance
957 self._printer = new_printer
959 def set_logger_class(
960 self,
961 logger_class: type[EzLogger] | EzLogger,
962 **init_kwargs: Any,
963 ) -> None:
964 """
965 Replace the current logger with a custom logger class or instance.
967 Allows users to override the default logger with a custom class that
968 inherits from EzLogger. The method preserves current
969 configuration values (level, rotation, retention, compression) unless
970 explicitly overridden in init_kwargs.
972 Args:
973 logger_class: Custom logger class inheriting from EzLogger,
974 or an already instantiated EzLogger instance
975 **init_kwargs: Optional initialization parameters for the logger
976 class. If not provided, current configuration values are used.
978 Raises:
979 TypeError: If logger_class is not a valid class or instance
980 ValidationError: If initialization parameters are invalid
981 FileOperationError: If file operations fail during logger creation
982 (may be raised by the logger class constructor)
984 Example:
985 >>> from ezpl import Ezpl, EzLogger
986 >>>
987 >>> class CustomLogger(EzLogger):
988 ... def info(self, message):
989 ... super().info(f"[CUSTOM LOG] {message}")
990 >>>
991 >>> ezpl = Ezpl()
992 >>> ezpl.set_logger_class(CustomLogger, log_file="custom.log")
993 >>> ezpl.get_logger().info("Test")
994 """
995 from .core.exceptions import ValidationError
997 # If it's already an instance, use it directly
998 if isinstance(logger_class, EzLogger):
999 new_logger = logger_class
1000 # If it's a class, instantiate it
1001 elif isinstance(logger_class, type):
1002 # Validate that it's a subclass of EzLogger
1003 if not issubclass(logger_class, EzLogger):
1004 raise TypeError(
1005 f"logger_class must be a subclass of {EzLogger.__name__}, "
1006 f"got {logger_class.__name__}"
1007 )
1009 # Preserve current configuration values if not provided
1010 current_level = (
1011 self._logger.level
1012 if hasattr(self, "_logger") and self._logger
1013 else self._config_manager.get_file_logger_level()
1014 )
1015 current_log_file = (
1016 self._log_file
1017 if hasattr(self, "_log_file")
1018 else self._config_manager.get_log_file()
1019 )
1020 current_rotation = (
1021 self._logger.rotation
1022 if hasattr(self, "_logger") and self._logger
1023 else self._config_manager.get_log_rotation()
1024 )
1025 current_retention = (
1026 self._logger.retention
1027 if hasattr(self, "_logger") and self._logger
1028 else self._config_manager.get_log_retention()
1029 )
1030 current_compression = (
1031 self._logger.compression
1032 if hasattr(self, "_logger") and self._logger
1033 else self._config_manager.get_log_compression()
1034 )
1035 current_backtrace = (
1036 self._logger.backtrace
1037 if hasattr(self, "_logger") and self._logger
1038 else self._config_manager.get_log_backtrace()
1039 )
1040 current_diagnose = (
1041 self._logger.diagnose
1042 if hasattr(self, "_logger") and self._logger
1043 else self._config_manager.get_log_diagnose()
1044 )
1046 # Merge kwargs with default values
1047 init_params = {
1048 "log_file": init_kwargs.pop("log_file", current_log_file),
1049 "level": init_kwargs.pop("level", current_level),
1050 "rotation": init_kwargs.pop("rotation", current_rotation),
1051 "retention": init_kwargs.pop("retention", current_retention),
1052 "compression": init_kwargs.pop("compression", current_compression),
1053 "backtrace": init_kwargs.pop("backtrace", current_backtrace),
1054 "diagnose": init_kwargs.pop("diagnose", current_diagnose),
1055 }
1056 init_params.update(init_kwargs)
1058 # Close previous logger before creating new one to avoid resource leaks
1059 try:
1060 if hasattr(self, "_logger") and self._logger:
1061 self._logger.close()
1062 except (EzplError, OSError, RuntimeError) as e:
1063 logger.error(f"Error while closing previous logger: {e}")
1065 # Create new instance
1066 try:
1067 new_logger = logger_class(**init_params)
1068 except Exception as e:
1069 raise ValidationError(
1070 f"Failed to initialize logger class {logger_class.__name__}: {e}",
1071 "logger_class",
1072 str(logger_class),
1073 ) from e
1074 else:
1075 raise TypeError(
1076 f"logger_class must be a class or an instance of {EzLogger.__name__}, "
1077 f"got {type(logger_class).__name__}"
1078 )
1080 # Replace the instance
1081 self._logger = new_logger
1083 # ///////////////////////////////////////////////////////////////
1084 # CONFIGURATION METHODS
1085 # ///////////////////////////////////////////////////////////////
1087 def reload_config(self) -> None:
1088 """
1089 Reload configuration from file and environment variables.
1091 This method reloads the configuration and reapplies it to handlers.
1092 Useful when environment variables or the config file have changed
1093 after the singleton was initialized.
1095 Note: This will reinitialize handlers with the new configuration.
1096 """
1097 # Reload configuration
1098 self._config_manager.reload()
1100 # Get configuration values
1101 printer_level = self._config_manager.get_printer_level()
1102 file_logger_level = self._config_manager.get_file_logger_level()
1103 global_log_level = self._config_manager.get_log_level()
1105 # Check if specific levels are explicitly set (not just defaults)
1106 # Priority: specific levels > global level
1107 # Only apply global level if specific levels are not explicitly set
1108 printer_level_explicit = self._config_manager.has_key("printer-level")
1109 file_logger_level_explicit = self._config_manager.has_key("file-logger-level")
1110 global_log_level_explicit = self._config_manager.has_key("log-level")
1112 # Respect manually set levels: don't override if set via set_level()
1113 printer_manually_set = self._printer.level_manually_set
1114 logger_manually_set = self._logger.level_manually_set
1116 # Reapply to handlers with priority logic (skip if manually set)
1117 if not printer_manually_set: 1117 ↛ 1118line 1117 didn't jump to line 1118 because the condition on line 1117 was never true
1118 if printer_level_explicit:
1119 effective_printer = printer_level
1120 elif global_log_level_explicit:
1121 effective_printer = global_log_level
1122 else:
1123 effective_printer = printer_level
1124 self.set_printer_level(effective_printer)
1126 if not logger_manually_set: 1126 ↛ 1127line 1126 didn't jump to line 1127 because the condition on line 1126 was never true
1127 if file_logger_level_explicit:
1128 effective_logger = file_logger_level
1129 elif global_log_level_explicit:
1130 effective_logger = global_log_level
1131 else:
1132 effective_logger = file_logger_level
1133 self.set_logger_level(effective_logger)
1135 # Reinitialize handlers — new instances have _level_manually_set = False by default
1136 self._rebuild_logger()
1137 self._rebuild_printer()
1139 def configure(
1140 self,
1141 config_dict: dict[str, Any] | None = None,
1142 *,
1143 persist: bool = False,
1144 **kwargs: Any,
1145 ) -> bool:
1146 """
1147 Configure Ezpl dynamically.
1149 Args:
1150 config_dict: Dictionary of configuration values to update
1151 persist: If True, write changes to ~/.ezpl/config.json so they
1152 survive future runs. Defaults to False (in-memory only).
1153 **kwargs: Configuration options (alternative to config_dict):
1154 - log_file or log-file: Path to log file
1155 - printer_level or printer-level: Printer log level
1156 - file_logger_level or file-logger-level: File logger level
1157 - level or log-level: Set both printer and logger level
1158 - log_rotation or log-rotation: Rotation setting (e.g., "10 MB", "1 day")
1159 - log_retention or log-retention: Retention period (e.g., "7 days")
1160 - log_compression or log-compression: Compression format (e.g., "zip", "gz")
1161 - log_backtrace or log-backtrace: Extend tracebacks beyond the
1162 catching point (default True)
1163 - log_diagnose or log-diagnose: Include local variable values in
1164 tracebacks (default False — enabling this may leak secrets
1165 into the log file)
1166 - indent_step or indent-step: Indentation step size
1167 - indent_symbol or indent-symbol: Symbol for indentation
1168 - base_indent_symbol or base-indent-symbol: Base indentation symbol
1170 Returns:
1171 True if configuration was applied, False if it was blocked by lock.
1172 """
1173 # Merge config_dict and kwargs
1174 if config_dict:
1175 kwargs.update(config_dict)
1177 # If configuration is locked, warn and return False
1178 if not self._can_write_config():
1179 warnings.warn(
1180 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) "
1181 "to unlock before reconfiguring.",
1182 UserWarning,
1183 stacklevel=2,
1184 )
1185 return False
1187 # Normalize keys: convert underscores to hyphens for consistency
1188 normalized_config = {}
1189 key_mapping = {
1190 "log_file": "log-file",
1191 "printer_level": "printer-level",
1192 "file_logger_level": "file-logger-level",
1193 "level": "log-level",
1194 "log_rotation": "log-rotation",
1195 "log_retention": "log-retention",
1196 "log_compression": "log-compression",
1197 "log_backtrace": "log-backtrace",
1198 "log_diagnose": "log-diagnose",
1199 "indent_step": "indent-step",
1200 "indent_symbol": "indent-symbol",
1201 "base_indent_symbol": "base-indent-symbol",
1202 }
1204 for key, value in kwargs.items():
1205 # Use normalized key if mapping exists, otherwise keep original
1206 normalized_key = key_mapping.get(key, key)
1207 normalized_config[normalized_key] = value
1209 # Extract log-file before update — set_log_file() owns that write.
1210 # Keep it persisted in config when requested.
1211 new_log_file = normalized_config.pop("log-file", None)
1212 if new_log_file is not None:
1213 normalized_config["log-file"] = str(Path(new_log_file))
1215 # Update configuration manager
1216 self._config_manager.update(normalized_config)
1217 # Apply changes to handlers
1218 if new_log_file is not None:
1219 self.set_log_file(new_log_file)
1221 # Handle log level changes with priority: specific > global
1222 self._apply_level_priority(
1223 printer_level=normalized_config.get("printer-level"),
1224 file_logger_level=normalized_config.get("file-logger-level"),
1225 global_level=normalized_config.get("log-level"),
1226 )
1228 # Reinitialize logger if rotation settings changed
1229 rotation_changed = any(
1230 key in normalized_config
1231 for key in [
1232 "log-rotation",
1233 "log-retention",
1234 "log-compression",
1235 "log-backtrace",
1236 "log-diagnose",
1237 ]
1238 )
1239 if rotation_changed:
1240 self._rebuild_logger()
1242 # Reinitialize printer if indent settings changed
1243 indent_changed = any(
1244 key in normalized_config
1245 for key in ["indent-step", "indent-symbol", "base-indent-symbol"]
1246 )
1247 if indent_changed: 1247 ↛ 1248line 1247 didn't jump to line 1248 because the condition on line 1247 was never true
1248 self._rebuild_printer()
1250 if persist:
1251 self._config_manager.save()
1253 return True
1256# ///////////////////////////////////////////////////////////////
1257# PUBLIC API
1258# ///////////////////////////////////////////////////////////////
1260__all__ = [
1261 "Ezpl",
1262]