Source code for config_wrangler.config_templates.logging_config

import logging
import sys
from typing import *
from contextlib import contextmanager
from datetime import datetime, time
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
from pathlib import Path

from pydantic import ByteSize, Field, model_validator
from pydicti import Dicti

from config_wrangler.config_templates.config_hierarchy import ConfigHierarchy
from config_wrangler.config_types.enum import StrEnum, auto_str
from config_wrangler.config_types.path_types import AutoCreateDirectoryPath
from config_wrangler.utils import TZFormatter

[docs] class LogLevel(StrEnum): CRITICAL = auto_str() FATAL = auto_str() ERROR = auto_str() WARNING = auto_str() INFO = auto_str() DEBUG = auto_str() NOTSET = auto_str()
[docs] class FileHandlerClass(StrEnum): # RotatingFileHandler = auto_str() # TimedRotatingFileHandler = auto_str()
[docs] class LoggingConfig(ConfigHierarchy): console_log_level: LogLevel = LogLevel.INFO console_entry_format: str = '%(asctime)s - %(levelname)-8s - %(name)s: %(message)s' log_folder: Optional[AutoCreateDirectoryPath] = None log_file_name: Optional[str] = None add_date_to_log_file_name: Optional[bool] = None log_file_name_date_time_format: str = '_%Y_%m_%d_at_%H_%M_%S' file_log_level: LogLevel = LogLevel.DEBUG log_file_entry_format: str = '%(asctime)s - %(levelname)-8s - %(name)s: %(message)s' log_file_rotation_class: FileHandlerClass = FileHandlerClass.RotatingFileHandler # RotatingFileHandler specific settings # log_file_max_size: ByteSize = Field(default='10 MB', validate_default=True) # TimedRotatingFileHandler specific settings # log_file_timed_rotation_when: str = 'midnight' log_file_timed_rotation_interval: int = 1 log_file_timed_rotation_attime: Optional[time] = None """ Note: If log_file_timed_rotation_attime is included it must be a valid time with or without seconds: 10:00 10:30 10:45:59 Note 2: The time validator will accept a value 0-59 but it is interpreted as seconds past midnight, not as hours like you might expect. """ log_file_timed_rotation_utc: bool = False log_files_to_keep: int = 10 logging_date_format: str = '%Y-%m-%d %H:%M:%S%z' trace_logging_setup: bool = False log_levels: Dict[str, LogLevel] @model_validator(mode='after') def _validate_logging(self): if self.log_file_name is not None: if self.log_folder is None: raise ValueError(f"{self.full_item_name()} log_file_name set but no log_folder provided") return self
[docs] @staticmethod def get_dated_log_file_name( log_file_prefix: str, date_time_format: str, log_file_suffix: str = '.log', ): """ Generates a log file name with a given prefix, suffix and date time format. Parameters ---------- log_file_prefix: str The part of the log file name before the date log_file_suffix: str The part of the log file name after the date date_time_format: str Optional. The date time format to use. Defaults to '_%Y_%m_%d_at_%H_%M_%S' """ return f"{log_file_prefix}{}{log_file_suffix}"
[docs] def add_log_file_handler( self, log_file_prefix: str = None, add_date_to_log_file_name: bool = None, log_file_suffix: str = '.log', ) -> logging.Handler: log = logging.getLogger(__name__) root_logger = logging.getLogger() if add_date_to_log_file_name is None: add_date_to_log_file_name = self.add_date_to_log_file_name if log_file_prefix is None: if self.log_file_name is not None: log_file_name = self.log_file_name else: log_file_name = None else:'Using code provided prefix {log_file_prefix}') if add_date_to_log_file_name is None: if self.log_file_rotation_class == FileHandlerClass.TimedRotatingFileHandler: add_date_to_log_file_name = False else: add_date_to_log_file_name = True if add_date_to_log_file_name: log_file_name = LoggingConfig.get_dated_log_file_name( log_file_prefix=log_file_prefix, date_time_format=self.log_file_name_date_time_format, log_file_suffix=log_file_suffix, ) else: log_file_name = f"{log_file_prefix}{log_file_suffix}" file_handler = None if log_file_name is not None: if self.log_folder is not None and not Path(log_file_name).is_absolute(): log_file_path = Path(self.log_folder, log_file_name) else: log_file_path = Path(log_file_name) if self.trace_logging_setup:'Logging path = {log_file_path}') # Setup file logging # Make sure the directory exists dir_name = log_file_path.parent if dir_name != '.': dir_name.mkdir(parents=True, exist_ok=True) if self.log_file_rotation_class == FileHandlerClass.RotatingFileHandler: file_handler = RotatingFileHandler( filename=log_file_path, maxBytes=self.log_file_max_size, backupCount=self.log_files_to_keep, encoding='utf8', ) elif self.log_file_rotation_class == FileHandlerClass.TimedRotatingFileHandler: file_handler = TimedRotatingFileHandler( filename=log_file_path, when=self.log_file_timed_rotation_when, interval=self.log_file_timed_rotation_interval, atTime=self.log_file_timed_rotation_attime, utc=self.log_file_timed_rotation_utc, backupCount=self.log_files_to_keep, encoding='utf8', ) file_handler.namer = lambda name: name.replace(".log", "") + ".log" else: raise ValueError(f"Bad log_file_rotation_class of {self.log_file_rotation_class}") log_file_entry_formatter = TZFormatter(self.log_file_entry_format, self.logging_date_format) file_handler.setFormatter(log_file_entry_formatter) file_handler.setLevel(self.file_log_level) root_logger.addHandler(file_handler)"File log level = {self.file_log_level}") else: if self.trace_logging_setup:'No log filename defined. File logging skipped.') return file_handler
[docs] def remove_log_handler(self, handler: logging.Handler): log = logging.getLogger(__name__) if handler is not None: if self.trace_logging_setup:"Closing log handler {handler}") root_logger = logging.getLogger() root_logger.removeHandler(handler) handler.close() else: if self.trace_logging_setup:'No handler provided to remove_log_handler')
[docs] @contextmanager def log_file_manager( self, log_file_prefix: str = None, add_date_to_log_file_name: bool = False, log_file_suffix: str = '.log', ): log_file_handler = self.add_log_file_handler( log_file_prefix=log_file_prefix, add_date_to_log_file_name=add_date_to_log_file_name, log_file_suffix=log_file_suffix, ) try: yield log_file_handler finally: # Code to release resource, e.g.: self.remove_log_handler(log_file_handler)
[docs] def setup_log_levels(self): configured_loggers = dict() root_logger = logging.getLogger() # Monkey-patch getLogger's dict to be case-insensitive logger_dict = Dicti(logging.Logger.manager.loggerDict) # @UndefinedVariable logging.Logger.manager.loggerDict = logger_dict for logger_class, desired_level_name in self.log_levels.items(): if logger_class.lower() == 'root': logger = root_logger else: logger = logging.getLogger(logger_class) configured_loggers[logger_class] = logger logger.propagate = True if desired_level_name is None: desired_level_name = 'INFO' else: desired_level_name = desired_level_name.upper() if self.trace_logging_setup: # Note: We can't use logging here yet print(f'Setting logger {} to {desired_level_name}') logger.setLevel(desired_level_name) # Will not include root logger for logger_class in sorted(logging.Logger.manager.loggerDict): logger = logging.getLogger(logger_class) if self.trace_logging_setup: print(f'Logger {logger_class} handlers {logger.handlers}') if logger_class not in configured_loggers: if self.trace_logging_setup: # Note: We can't use logging here yet print( f"Checking existing logger {logger_class} " f"level {logging.getLevelName(logger.getEffectiveLevel())}" ) for compare_logger in sorted(configured_loggers): if logger_class.startswith(compare_logger): parent_logger = configured_loggers[compare_logger] level = parent_logger.getEffectiveLevel() logger.setLevel(level) if self.trace_logging_setup: # Note: We can't use logging here yet print( f"Existing logger {logger_class} re-setup with {} " f"settings {logging.getLevelName(level)}" ) logger.propagate = True if self.trace_logging_setup: # Note: We can't use logging here yet print(f"Root logging level is {logging.getLevelName(root_logger.getEffectiveLevel())}") print(f"Root logging handlers are {root_logger.handlers}")
[docs] def setup_logging( self, log_file_prefix: str = None, add_date_to_log_file_name: bool = None, log_file_suffix: str = '.log', console_output=None, use_log_file_setting=True, ): """ Setup logging based on configuration. """ root_logger = logging.getLogger() if not hasattr(root_logger, 'config_wrangler_setup_done'): # Store info that we have already setup logging in the root logger. # This could be stored in the config, it would be too easy for methods to open # their own configs and thus will not know the logging had already been set up. root_logger.config_wrangler_setup_done = True # Close out any existing handlers for handler in root_logger.handlers: handler.flush() handler.close() # Reset the handlers root_logger.handlers.clear() if self.trace_logging_setup: # Note: We can't use logging here yet print('logging.trace_setup is True}') print(f'Starting root logger handlers {root_logger.handlers}') # Modify the root logger level to at least INFO for now. setup_log_levels could change it root_logger.setLevel(logging.INFO) if console_output is None: error_output = sys.stderr regular_output = sys.stdout else: error_output = console_output regular_output = console_output console_error_log = logging.StreamHandler(error_output) console_error_log.setLevel(logging.ERROR) root_logger.addHandler(console_error_log) console_log = logging.StreamHandler(regular_output) console_log.setLevel(self.console_log_level) # define a filter for non-error messages def non_error(record): return record.levelno != logging.ERROR # Errors go to console_error_log above, so we don't want them here as well console_log.addFilter(non_error) root_logger.addHandler(console_log) console_entry_format = self.console_entry_format if console_entry_format: console_entry_formatter = TZFormatter(console_entry_format, self.logging_date_format) console_log.setFormatter(console_entry_formatter) console_error_log.setFormatter(console_entry_formatter) self.setup_log_levels() # Switch to this modules logger log = logging.getLogger(__name__) logging.captureWarnings(True) log_level_name = logging.getLevelName(log.getEffectiveLevel()) if self.trace_logging_setup:"This modules logging level is {log_level_name}") if use_log_file_setting or log_file_prefix is not None: return self.add_log_file_handler( log_file_prefix=log_file_prefix, add_date_to_log_file_name=add_date_to_log_file_name, log_file_suffix=log_file_suffix, ) else: if self.trace_logging_setup:'use_log_file_setting = False. setup_log_file not called.') return None