Coverage for src / ezplog / ezpl.py: 67.15%
268 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-30 19:43 +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
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 *,
105 lock_config: bool = False,
106 intercept_stdlib: bool = False,
107 ) -> Ezpl:
108 """
109 Creates and returns a new instance of Ezpl if none exists.
111 **Notes:**
112 Ensures only one instance of Ezpl exists (Singleton pattern).
114 **Priority order for configuration (for each parameter):**
115 1. Arguments passed directly (highest priority)
116 2. Environment variables (EZPL_*)
117 3. Configuration file (~/.ezpl/config.json)
118 4. Default values (lowest priority)
120 **Args:**
122 * `log_file` (Path | str, optional): Path to the log file
123 * `log_level` (str, optional): Global log level (applies to both printer and logger)
124 * `printer_level` (str, optional): Printer log level
125 * `file_logger_level` (str, optional): File logger level
126 * `log_rotation` (str, optional): Rotation setting (e.g., "10 MB", "1 day")
127 * `log_retention` (str, optional): Retention period (e.g., "7 days")
128 * `log_compression` (str, optional): Compression format (e.g., "zip", "gz")
129 * `indent_step` (int, optional): Indentation step size
130 * `indent_symbol` (str, optional): Symbol for indentation
131 * `base_indent_symbol` (str, optional): Base indentation symbol
132 * `lock_config` (bool): If True, lock configuration immediately after init.
133 Subsequent configure() calls will be blocked unless a valid token is used.
134 The lock token is stored in Ezpl._config_lock_token.
135 * `intercept_stdlib` (bool): If True, install an InterceptHandler on the root
136 stdlib logger so that all library loggers using logging.getLogger(__name__)
137 are automatically forwarded to the loguru pipeline.
139 **Returns:**
141 * `Ezpl`: The singleton instance of the Ezpl class.
143 **Raises:**
145 * `None`.
146 """
148 # //////
149 # Double-checked locking pattern for thread-safe singleton
150 if cls._instance is None:
151 with cls._lock:
152 # Check again after acquiring lock (double-checked locking)
153 if cls._instance is None:
154 # Initialize configuration manager
155 cls._config_manager = ConfigurationManager()
157 # Determine configuration values with priority: arg > env > config file > default
158 # Helper function to get value with priority order
159 def get_config_value(
160 arg_value, config_key: str, getter_method
161 ) -> Any:
162 """
163 Get configuration value with priority: arg > env > config file > default
165 Args:
166 arg_value: Value from argument (can be None)
167 config_key: Configuration key name
168 getter_method: Method to get default value from config manager
170 Returns:
171 Final configuration value
172 """
173 # Priority 1: Argument direct
174 if arg_value is not None:
175 return arg_value
177 # Priority 2: Environment variable (already loaded in config_manager)
178 # Priority 3: Config file (already loaded in config_manager)
179 # Priority 4: Default (via getter method)
180 # The config_manager already has the correct priority (env > file > default)
181 config_value = cls._config_manager.get(config_key)
182 if config_value is not None: 182 ↛ 186line 182 didn't jump to line 186 because the condition on line 182 was always true
183 return config_value
185 # Fallback to default via getter
186 return getter_method()
188 # Log file
189 if log_file:
190 cls._log_file = Path(log_file)
191 else:
192 cls._log_file = cls._config_manager.get_log_file()
194 # Log level (global)
195 final_log_level = get_config_value(
196 log_level, "log-level", cls._config_manager.get_log_level
197 )
199 # Printer level
200 final_printer_level = get_config_value(
201 printer_level,
202 "printer-level",
203 cls._config_manager.get_printer_level,
204 )
206 # File logger level
207 final_file_logger_level = get_config_value(
208 file_logger_level,
209 "file-logger-level",
210 cls._config_manager.get_file_logger_level,
211 )
213 # Rotation settings (can be None)
214 # Priority: arg > env > config file > default
215 # Note: If arg is None (default), we check env/config/default
216 # If user wants to explicitly set None, they can pass None or use configure()
217 final_rotation = (
218 log_rotation
219 if log_rotation is not None
220 else cls._config_manager.get_log_rotation()
221 )
222 final_retention = (
223 log_retention
224 if log_retention is not None
225 else cls._config_manager.get_log_retention()
226 )
227 final_compression = (
228 log_compression
229 if log_compression is not None
230 else cls._config_manager.get_log_compression()
231 )
233 # Indent settings
234 final_indent_step = get_config_value(
235 indent_step, "indent-step", cls._config_manager.get_indent_step
236 )
237 final_indent_symbol = get_config_value(
238 indent_symbol,
239 "indent-symbol",
240 cls._config_manager.get_indent_symbol,
241 )
242 final_base_indent_symbol = get_config_value(
243 base_indent_symbol,
244 "base-indent-symbol",
245 cls._config_manager.get_base_indent_symbol,
246 )
248 instance = object.__new__(cls)
249 cls._instance = instance
251 # Initialize printer with resolved configuration
252 cls._printer = EzPrinter(
253 level=final_printer_level,
254 indent_step=final_indent_step,
255 indent_symbol=final_indent_symbol,
256 base_indent_symbol=final_base_indent_symbol,
257 )
259 # Initialize logger with resolved configuration
260 cls._logger = EzLogger(
261 log_file=cls._log_file,
262 level=final_file_logger_level,
263 rotation=final_rotation,
264 retention=final_retention,
265 compression=final_compression,
266 )
268 # Apply global log level with priority: specific > global
269 cls._instance._apply_level_priority(
270 printer_level=printer_level,
271 file_logger_level=file_logger_level,
272 global_level=final_log_level,
273 )
275 # Apply post-init options — inside the critical section,
276 # first construction only (double-checked locking guarantees this)
277 if lock_config: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true
278 cls._config_lock_token = cls.lock_config()
280 if intercept_stdlib: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true
281 cls._install_intercept_handler()
283 # Type narrowing: _instance is guaranteed to be set at this point
284 if cls._instance is None: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 raise InitializationError("Singleton initialization failed unexpectedly")
286 return cls._instance
288 # ///////////////////////////////////////////////////////////////
289 # PRIVATE HELPERS
290 # ///////////////////////////////////////////////////////////////
292 def _apply_level_priority(
293 self,
294 *,
295 printer_level: str | None = None,
296 file_logger_level: str | None = None,
297 global_level: str | None = None,
298 ) -> None:
299 """
300 Apply log levels with priority: specific level > global level.
302 Only sets levels when a non-None value can be resolved.
303 If a specific level is not provided, global_level is used as fallback.
304 """
305 effective_printer = printer_level if printer_level is not None else global_level
306 effective_logger = (
307 file_logger_level if file_logger_level is not None else global_level
308 )
310 if effective_printer:
311 self.set_printer_level(effective_printer)
313 if effective_logger:
314 self.set_logger_level(effective_logger)
316 def _rebuild_logger(
317 self,
318 *,
319 level: str | None = None,
320 ) -> None:
321 """
322 Close the current file logger and reinitialize it with current configuration.
324 Args:
325 level: Override the log level. If None, preserves the current level.
326 """
327 current_level = (
328 level
329 or (
330 self._logger.level
331 if hasattr(self, "_logger") and self._logger
332 else None
333 )
334 or self._config_manager.get_file_logger_level()
335 )
336 try:
337 if hasattr(self, "_logger") and self._logger: 337 ↛ 341line 337 didn't jump to line 341 because the condition on line 337 was always true
338 self._logger.close()
339 except (EzplError, OSError, RuntimeError) as e:
340 logger.error(f"Error while closing logger: {e}")
341 self._logger = EzLogger(
342 log_file=self._log_file,
343 level=current_level,
344 rotation=self._config_manager.get_log_rotation(),
345 retention=self._config_manager.get_log_retention(),
346 compression=self._config_manager.get_log_compression(),
347 )
349 def _rebuild_printer(
350 self,
351 *,
352 level: str | None = None,
353 ) -> None:
354 """
355 Reinitialize the console printer with current configuration.
357 Args:
358 level: Override the log level. If None, preserves the current level.
359 """
360 current_level = (
361 level
362 or (
363 self._printer.level
364 if hasattr(self, "_printer") and self._printer
365 else None
366 )
367 or self._config_manager.get_printer_level()
368 )
369 self._printer = EzPrinter(
370 level=current_level,
371 indent_step=self._config_manager.get_indent_step(),
372 indent_symbol=self._config_manager.get_indent_symbol(),
373 base_indent_symbol=self._config_manager.get_base_indent_symbol(),
374 )
376 # ///////////////////////////////////////////////////////////////
377 # GETTER
378 # ///////////////////////////////////////////////////////////////
380 def get_printer(self) -> EzPrinter:
381 """
382 Returns the EzPrinter instance.
384 **Returns:**
386 * EzPrinter: The console printer instance providing info(), debug(), success(), etc.
387 Implements PrinterProtocol for type safety.
388 """
389 return self._printer
391 # ///////////////////////////////////////////////////////////////
393 def get_logger(self) -> EzLogger:
394 """
395 Returns the EzLogger instance.
397 **Returns:**
399 * EzLogger: The file logger instance for file logging.
400 Use logger.info(), logger.debug(), etc. directly.
401 For advanced loguru features, use logger.get_loguru()
402 Implements LoggerProtocol for type safety.
403 """
404 return self._logger
406 # ///////////////////////////////////////////////////////////////
407 # UTILS METHODS
408 # ///////////////////////////////////////////////////////////////
410 @property
411 def printer_level(self) -> str:
412 """Return the current printer logging level."""
413 return self._printer.level
415 @property
416 def logger_level(self) -> str:
417 """Return the current file logger logging level."""
418 return self._logger.level
420 def set_level(self, level: str) -> None:
421 """
422 Set the log level for both the printer and the logger simultaneously.
424 **Args:**
426 * `level` (str): The desired log level (e.g., "INFO", "WARNING").
428 **Returns:**
430 * `None`.
431 """
432 if not self._can_write_config(): 432 ↛ 433line 432 didn't jump to line 433 because the condition on line 432 was never true
433 warnings.warn(
434 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock.",
435 UserWarning,
436 stacklevel=2,
437 )
438 return
439 self.set_logger_level(level)
440 self.set_printer_level(level)
442 def set_printer_level(self, level: str) -> None:
443 """
444 Set the log level for the printer only.
446 **Args:**
448 * `level` (str): The desired log level for the printer.
450 **Returns:**
452 * `None`.
453 """
454 if not self._can_write_config(): 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 warnings.warn(
456 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) to unlock.",
457 UserWarning,
458 stacklevel=2,
459 )
460 return
461 self._printer.set_level(level)
463 def set_logger_level(self, level: str) -> None:
464 """
465 Set the log level for the file logger only.
467 **Args:**
469 * `level` (str): The desired log level for the file logger.
471 **Returns:**
473 * `None`.
474 """
475 if not self._can_write_config(): 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(token) to unlock.",
478 UserWarning,
479 stacklevel=2,
480 )
481 return
482 self._logger.set_level(level)
484 # ///////////////////////////////////////////////////////////////
485 # FACADE METHODS
486 # ///////////////////////////////////////////////////////////////
488 def debug(self, message: Any) -> None:
489 """Log a debug message to the console printer."""
490 self._printer.debug(message)
492 def info(self, message: Any) -> None:
493 """Log an info message to the console printer."""
494 self._printer.info(message)
496 def success(self, message: Any) -> None:
497 """Log a success message to the console printer."""
498 self._printer.success(message)
500 def warning(self, message: Any) -> None:
501 """Log a warning message to the console printer."""
502 self._printer.warning(message)
504 def error(self, message: Any) -> None:
505 """Log an error message to the console printer."""
506 self._printer.error(message)
508 def critical(self, message: Any) -> None:
509 """Log a critical message to the console printer."""
510 self._printer.critical(message)
512 # ///////////////////////////////////////////////////////////////
514 def add_separator(self) -> None:
515 """
516 Adds a separator to the log file.
518 **Returns:**
520 * `None`.
521 """
522 self._logger.add_separator()
524 # ///////////////////////////////////////////////////////////////
526 @contextmanager
527 def manage_indent(self) -> Generator[None, None, None]:
528 """
529 Context manager to manage indentation level.
531 **Returns:**
533 * `None`.
534 """
535 with self._printer.manage_indent():
536 yield
538 # ///////////////////////////////////////////////////////////////
539 # ENHANCED METHODS
540 # ///////////////////////////////////////////////////////////////
542 @classmethod
543 def reset(cls) -> None:
544 """
545 Reset the singleton instance (useful for testing).
547 Warning: This will destroy the current instance and all its state.
548 """
549 if cls._instance is not None:
550 # Close logger handlers to release file handles (important on Windows)
551 try:
552 if hasattr(cls._instance, "_logger") and cls._instance._logger: 552 ↛ 556line 552 didn't jump to line 556 because the condition on line 552 was always true
553 cls._instance._logger.close()
554 except (EzplError, OSError, RuntimeError) as e:
555 logger.error(f"Error during cleanup: {e}")
556 cls._instance = None
557 # Also reset configuration lock
558 cls._config_locked = False
559 cls._config_lock_token = None
561 # ------------------------------------------------
562 # CONFIG LOCK CONTROL
563 # ------------------------------------------------
565 @classmethod
566 def lock_config(cls) -> str:
567 """
568 Lock Ezpl configuration so that future configure() and set_level() calls
569 are blocked until unlock_config(token) is called.
571 Intended usage:
572 1. Root application configures Ezpl once
573 2. Calls token = Ezpl.lock_config()
574 3. Stores the token; libraries cannot reconfigure Ezpl without it
576 Returns:
577 str: A token required to unlock the configuration later.
578 """
579 with cls._lock:
580 if not cls._config_locked:
581 cls._config_locked = True
582 cls._config_lock_token = uuid4().hex
583 token = cls._config_lock_token
584 if token is None: 584 ↛ 586line 584 didn't jump to line 586 because the condition on line 584 was never true
585 # Should never happen: token is set whenever _config_locked is True
586 raise RuntimeError(
587 "Configuration lock token is None despite lock being active"
588 )
589 return token
591 @classmethod
592 def unlock_config(cls, token: str) -> bool:
593 """
594 Unlock Ezpl configuration.
596 Args:
597 token: The token returned by lock_config(). Must match exactly.
599 Returns:
600 True if unlocked successfully, False if the token is wrong.
601 """
602 with cls._lock:
603 if not cls._config_locked: 603 ↛ 604line 603 didn't jump to line 604 because the condition on line 603 was never true
604 return True
606 if token != cls._config_lock_token:
607 warnings.warn(
608 "Unlock denied: invalid token.",
609 UserWarning,
610 stacklevel=2,
611 )
612 return False
614 cls._config_locked = False
615 cls._config_lock_token = None
616 return True
618 @classmethod
619 def is_locked(cls) -> bool:
620 """Return True if the configuration is currently locked."""
621 return cls._config_locked
623 @classmethod
624 def _can_write_config(cls) -> bool:
625 """Return True when configuration writes are allowed."""
626 return not cls._config_locked
628 @classmethod
629 def _install_intercept_handler(cls) -> None:
630 """
631 Install InterceptHandler on the root stdlib logger.
633 After this call, all records emitted via logging.getLogger(__name__)
634 — including those from ezpl.lib_mode.get_logger() — are forwarded
635 to the loguru pipeline.
637 This method is idempotent: calling it multiple times installs the
638 handler only once.
639 """
640 import logging as _logging
642 root = _logging.getLogger()
643 if not any(isinstance(h, InterceptHandler) for h in root.handlers):
644 root.addHandler(InterceptHandler())
645 root.setLevel(0)
647 def set_log_file(self, log_file: Path | str) -> None:
648 """
649 Change the log file (requires reinitialization of the logger).
651 Args:
652 log_file: New path to the log file
654 Note: This will reinitialize the file logger but keep the singleton instance.
655 """
656 new_log_file = Path(log_file)
657 if new_log_file != self._log_file: 657 ↛ exitline 657 didn't return from function 'set_log_file' because the condition on line 657 was always true
658 self._log_file = new_log_file
659 # Update configuration
660 self._config_manager.set("log-file", str(new_log_file))
661 self._rebuild_logger()
663 def get_log_file(self) -> Path:
664 """
665 Get the current log file path.
667 Returns:
668 Path to the current log file
669 """
670 return self._log_file
672 def get_config(self) -> ConfigurationManager:
673 """
674 Get the current configuration manager.
676 Returns:
677 ConfigurationManager instance for accessing and modifying configuration
678 """
679 return self._config_manager
681 # ///////////////////////////////////////////////////////////////
682 # HANDLER OVERRIDE METHODS
683 # ///////////////////////////////////////////////////////////////
685 def set_printer_class(
686 self,
687 printer_class: type[EzPrinter] | EzPrinter,
688 **init_kwargs: Any,
689 ) -> None:
690 """
691 Replace the current printer with a custom printer class or instance.
693 Allows users to override the default printer with a custom class that
694 inherits from EzPrinter. The method preserves
695 current configuration values (level, indentation settings) unless
696 explicitly overridden in init_kwargs.
698 Args:
699 printer_class: Custom printer class inheriting from EzPrinter,
700 or an already instantiated EzPrinter instance
701 **init_kwargs: Optional initialization parameters for the printer
702 class. If not provided, current configuration values are used.
704 Raises:
705 TypeError: If printer_class is not a valid class or instance
706 ValidationError: If initialization parameters are invalid
708 Example:
709 >>> from ezpl import Ezpl, EzPrinter
710 >>>
711 >>> class CustomPrinter(EzPrinter):
712 ... def info(self, message):
713 ... super().info(f"[CUSTOM] {message}")
714 >>>
715 >>> ezpl = Ezpl()
716 >>> ezpl.set_printer_class(CustomPrinter, level="DEBUG")
717 >>> ezpl.get_printer().info("Test")
718 [CUSTOM] Test
719 """
720 from .core.exceptions import ValidationError
722 # If it's already an instance, use it directly
723 if isinstance(printer_class, EzPrinter):
724 new_printer = printer_class
725 # If it's a class, instantiate it
726 elif isinstance(printer_class, type):
727 # Validate that it's a subclass of EzPrinter
728 if not issubclass(printer_class, EzPrinter):
729 raise TypeError(
730 f"printer_class must be a subclass of {EzPrinter.__name__}, "
731 f"got {printer_class.__name__}"
732 )
734 # Preserve current configuration values if not provided
735 current_level = (
736 self._printer.level
737 if hasattr(self, "_printer") and self._printer
738 else self._config_manager.get_printer_level()
739 )
740 current_indent_step = (
741 self._printer.indent_step
742 if hasattr(self, "_printer") and self._printer
743 else self._config_manager.get_indent_step()
744 )
745 current_indent_symbol = (
746 self._printer.indent_symbol
747 if hasattr(self, "_printer") and self._printer
748 else self._config_manager.get_indent_symbol()
749 )
750 current_base_indent_symbol = (
751 self._printer.base_indent_symbol
752 if hasattr(self, "_printer") and self._printer
753 else self._config_manager.get_base_indent_symbol()
754 )
756 # Merge kwargs with default values
757 init_params = {
758 "level": init_kwargs.pop("level", current_level),
759 "indent_step": init_kwargs.pop("indent_step", current_indent_step),
760 "indent_symbol": init_kwargs.pop(
761 "indent_symbol", current_indent_symbol
762 ),
763 "base_indent_symbol": init_kwargs.pop(
764 "base_indent_symbol", current_base_indent_symbol
765 ),
766 }
767 init_params.update(init_kwargs)
769 # Create new instance
770 try:
771 new_printer = printer_class(**init_params)
772 except Exception as e:
773 raise ValidationError(
774 f"Failed to initialize printer class {printer_class.__name__}: {e}",
775 "printer_class",
776 str(printer_class),
777 ) from e
778 else:
779 raise TypeError(
780 f"printer_class must be a class or an instance of {EzPrinter.__name__}, "
781 f"got {type(printer_class).__name__}"
782 )
784 # Replace the instance
785 self._printer = new_printer
787 def set_logger_class(
788 self,
789 logger_class: type[EzLogger] | EzLogger,
790 **init_kwargs: Any,
791 ) -> None:
792 """
793 Replace the current logger with a custom logger class or instance.
795 Allows users to override the default logger with a custom class that
796 inherits from EzLogger. The method preserves current
797 configuration values (level, rotation, retention, compression) unless
798 explicitly overridden in init_kwargs.
800 Args:
801 logger_class: Custom logger class inheriting from EzLogger,
802 or an already instantiated EzLogger instance
803 **init_kwargs: Optional initialization parameters for the logger
804 class. If not provided, current configuration values are used.
806 Raises:
807 TypeError: If logger_class is not a valid class or instance
808 ValidationError: If initialization parameters are invalid
809 FileOperationError: If file operations fail during logger creation
810 (may be raised by the logger class constructor)
812 Example:
813 >>> from ezpl import Ezpl, EzLogger
814 >>>
815 >>> class CustomLogger(EzLogger):
816 ... def info(self, message):
817 ... super().info(f"[CUSTOM LOG] {message}")
818 >>>
819 >>> ezpl = Ezpl()
820 >>> ezpl.set_logger_class(CustomLogger, log_file="custom.log")
821 >>> ezpl.get_logger().info("Test")
822 """
823 from .core.exceptions import ValidationError
825 # If it's already an instance, use it directly
826 if isinstance(logger_class, EzLogger):
827 new_logger = logger_class
828 # If it's a class, instantiate it
829 elif isinstance(logger_class, type):
830 # Validate that it's a subclass of EzLogger
831 if not issubclass(logger_class, EzLogger):
832 raise TypeError(
833 f"logger_class must be a subclass of {EzLogger.__name__}, "
834 f"got {logger_class.__name__}"
835 )
837 # Preserve current configuration values if not provided
838 current_level = (
839 self._logger.level
840 if hasattr(self, "_logger") and self._logger
841 else self._config_manager.get_file_logger_level()
842 )
843 current_log_file = (
844 self._log_file
845 if hasattr(self, "_log_file")
846 else self._config_manager.get_log_file()
847 )
848 current_rotation = (
849 self._logger.rotation
850 if hasattr(self, "_logger") and self._logger
851 else self._config_manager.get_log_rotation()
852 )
853 current_retention = (
854 self._logger.retention
855 if hasattr(self, "_logger") and self._logger
856 else self._config_manager.get_log_retention()
857 )
858 current_compression = (
859 self._logger.compression
860 if hasattr(self, "_logger") and self._logger
861 else self._config_manager.get_log_compression()
862 )
864 # Merge kwargs with default values
865 init_params = {
866 "log_file": init_kwargs.pop("log_file", current_log_file),
867 "level": init_kwargs.pop("level", current_level),
868 "rotation": init_kwargs.pop("rotation", current_rotation),
869 "retention": init_kwargs.pop("retention", current_retention),
870 "compression": init_kwargs.pop("compression", current_compression),
871 }
872 init_params.update(init_kwargs)
874 # Close previous logger before creating new one to avoid resource leaks
875 try:
876 if hasattr(self, "_logger") and self._logger:
877 self._logger.close()
878 except (EzplError, OSError, RuntimeError) as e:
879 logger.error(f"Error while closing previous logger: {e}")
881 # Create new instance
882 try:
883 new_logger = logger_class(**init_params)
884 except Exception as e:
885 raise ValidationError(
886 f"Failed to initialize logger class {logger_class.__name__}: {e}",
887 "logger_class",
888 str(logger_class),
889 ) from e
890 else:
891 raise TypeError(
892 f"logger_class must be a class or an instance of {EzLogger.__name__}, "
893 f"got {type(logger_class).__name__}"
894 )
896 # Replace the instance
897 self._logger = new_logger
899 # ///////////////////////////////////////////////////////////////
900 # CONFIGURATION METHODS
901 # ///////////////////////////////////////////////////////////////
903 def reload_config(self) -> None:
904 """
905 Reload configuration from file and environment variables.
907 This method reloads the configuration and reapplies it to handlers.
908 Useful when environment variables or the config file have changed
909 after the singleton was initialized.
911 Note: This will reinitialize handlers with the new configuration.
912 """
913 # Reload configuration
914 self._config_manager.reload()
916 # Get configuration values
917 printer_level = self._config_manager.get_printer_level()
918 file_logger_level = self._config_manager.get_file_logger_level()
919 global_log_level = self._config_manager.get_log_level()
921 # Check if specific levels are explicitly set (not just defaults)
922 # Priority: specific levels > global level
923 # Only apply global level if specific levels are not explicitly set
924 printer_level_explicit = self._config_manager.has_key("printer-level")
925 file_logger_level_explicit = self._config_manager.has_key("file-logger-level")
926 global_log_level_explicit = self._config_manager.has_key("log-level")
928 # Respect manually set levels: don't override if set via set_level()
929 printer_manually_set = self._printer.level_manually_set
930 logger_manually_set = self._logger.level_manually_set
932 # Reapply to handlers with priority logic (skip if manually set)
933 if not printer_manually_set: 933 ↛ 934line 933 didn't jump to line 934 because the condition on line 933 was never true
934 effective_printer = (
935 printer_level
936 if printer_level_explicit
937 else global_log_level
938 if global_log_level_explicit
939 else printer_level
940 )
941 self.set_printer_level(effective_printer)
943 if not logger_manually_set: 943 ↛ 944line 943 didn't jump to line 944 because the condition on line 943 was never true
944 effective_logger = (
945 file_logger_level
946 if file_logger_level_explicit
947 else (
948 global_log_level if global_log_level_explicit else file_logger_level
949 )
950 )
951 self.set_logger_level(effective_logger)
953 # Reinitialize handlers — new instances have _level_manually_set = False by default
954 self._rebuild_logger()
955 self._rebuild_printer()
957 def configure(
958 self,
959 config_dict: dict[str, Any] | None = None,
960 *,
961 persist: bool = False,
962 **kwargs: Any,
963 ) -> bool:
964 """
965 Configure Ezpl dynamically.
967 Args:
968 config_dict: Dictionary of configuration values to update
969 persist: If True, write changes to ~/.ezpl/config.json so they
970 survive future runs. Defaults to False (in-memory only).
971 **kwargs: Configuration options (alternative to config_dict):
972 - log_file or log-file: Path to log file
973 - printer_level or printer-level: Printer log level
974 - logger_level or file-logger-level: File logger level
975 - file_logger_level or file-logger-level: File logger level
976 - level or log-level: Set both printer and logger level
977 - log_rotation or log-rotation: Rotation setting (e.g., "10 MB", "1 day")
978 - log_retention or log-retention: Retention period (e.g., "7 days")
979 - log_compression or log-compression: Compression format (e.g., "zip", "gz")
980 - indent_step or indent-step: Indentation step size
981 - indent_symbol or indent-symbol: Symbol for indentation
982 - base_indent_symbol or base-indent-symbol: Base indentation symbol
984 Returns:
985 True if configuration was applied, False if it was blocked by lock.
986 """
987 # Merge config_dict and kwargs
988 if config_dict:
989 kwargs.update(config_dict)
991 # If configuration is locked, warn and return False
992 if not self._can_write_config():
993 warnings.warn(
994 "Ezpl configuration is locked. Call Ezpl.unlock_config(token) "
995 "to unlock before reconfiguring.",
996 UserWarning,
997 stacklevel=2,
998 )
999 return False
1001 # Normalize keys: convert underscores to hyphens for consistency
1002 normalized_config = {}
1003 key_mapping = {
1004 "log_file": "log-file",
1005 "printer_level": "printer-level",
1006 "logger_level": "file-logger-level",
1007 "file_logger_level": "file-logger-level",
1008 "level": "log-level",
1009 "log_rotation": "log-rotation",
1010 "log_retention": "log-retention",
1011 "log_compression": "log-compression",
1012 "indent_step": "indent-step",
1013 "indent_symbol": "indent-symbol",
1014 "base_indent_symbol": "base-indent-symbol",
1015 }
1017 for key, value in kwargs.items():
1018 # Use normalized key if mapping exists, otherwise keep original
1019 normalized_key = key_mapping.get(key, key)
1020 normalized_config[normalized_key] = value
1022 # Extract log-file before update — set_log_file() owns that write
1023 new_log_file = normalized_config.pop("log-file", None)
1025 # Update configuration manager
1026 self._config_manager.update(normalized_config)
1027 if persist: 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 self._config_manager.save()
1030 # Apply changes to handlers
1031 if new_log_file is not None: 1031 ↛ 1032line 1031 didn't jump to line 1032 because the condition on line 1031 was never true
1032 self.set_log_file(new_log_file)
1034 # Handle log level changes with priority: specific > global
1035 self._apply_level_priority(
1036 printer_level=normalized_config.get("printer-level"),
1037 file_logger_level=normalized_config.get("file-logger-level"),
1038 global_level=normalized_config.get("log-level"),
1039 )
1041 # Reinitialize logger if rotation settings changed
1042 rotation_changed = any(
1043 key in normalized_config
1044 for key in ["log-rotation", "log-retention", "log-compression"]
1045 )
1046 if rotation_changed:
1047 self._rebuild_logger()
1049 # Reinitialize printer if indent settings changed
1050 indent_changed = any(
1051 key in normalized_config
1052 for key in ["indent-step", "indent-symbol", "base-indent-symbol"]
1053 )
1054 if indent_changed: 1054 ↛ 1055line 1054 didn't jump to line 1055 because the condition on line 1054 was never true
1055 self._rebuild_printer()
1057 return True
1060# ///////////////////////////////////////////////////////////////
1061# PUBLIC API
1062# ///////////////////////////////////////////////////////////////
1064__all__ = [
1065 "Ezpl",
1066]