Source code for highspot.utils.log_utils

# -*- coding: utf-8 -*-
:Module:            highspot.utils.log_utils
:Synopsis:          Collection of logging utilities and functions
:Usage:             ``from highspot.utils import log_utils``
:Example:           ``logger = log_utils.initialize_logging(__name__)``
:Created By:        Jeff Shurtliff
:Last Modified:     Jeff Shurtliff
:Modified Date:     16 Oct 2022

import os
import sys
import logging
import logging.handlers
from pathlib import Path

    'logger_name': __name__,
    'log_level': 'info',
    'formatter': logging.Formatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s'),
    'date_format': '%Y-%m-%d %I:%M:%S',
    'file_log_level': 'info',
    'console_log_level': 'warning',
    'syslog_log_level': 'info',
    'syslog_address': 'localhost',
    'syslog_port': 514,

[docs]def initialize_logging(logger_name=None, log_level=None, formatter=None, debug=None, no_output=None, file_output=None, file_log_level=None, log_file=None, overwrite_log_files=None, console_output=None, console_log_level=None, syslog_output=None, syslog_log_level=None, syslog_address=None, syslog_port=None): """This function initializes logging for the highspot library.""" # TODO: Complete the docstring above with parameters logger_name, log_levels, formatter = _apply_defaults(logger_name, formatter, debug, log_level, file_log_level, console_log_level, syslog_log_level) log_level, file_log_level, console_log_level, syslog_log_level = _get_log_levels_from_dict(log_levels) logger = logging.getLogger(logger_name) logger = _set_logging_level(logger, log_level) logger = _add_handlers(logger, formatter, no_output, file_output, file_log_level, log_file, overwrite_log_files, console_output, console_log_level, syslog_output, syslog_log_level, syslog_address, syslog_port) return logger
[docs]class LessThanFilter(logging.Filter): """This class allows filters to be set to limit log levels to only less than a specified level. .. seealso:: `Zoey Greer <>`_ is the original author of this class which was provided on `Stack Overflow <>`_. """ def __init__(self, exclusive_maximum, name=""): """This method instantiates the :py:class:`highspot.utils.log_utils.LessThanFilter` class object.""" super(LessThanFilter, self).__init__(name) self.max_level = exclusive_maximum
[docs] def filter(self, record): """This method returns a Boolean integer value indicating whether or not a message should be logged. .. note:: A non-zero return indicates that the message will be logged. """ return 1 if record.levelno < self.max_level else 0
def _apply_defaults(_logger_name, _formatter, _debug, _log_level, _file_level, _console_level, _syslog_level): """This function applies default values to the configuration settings if not explicitly defined. :param _logger_name: The name of the logger instance :type _logger_name: str, None :param _formatter: The log format to utilize for the logger instance :type _formatter: str, None :param _debug: Defines if debug mode is enabled :type _debug: bool, None :param _log_level: The general logging level for the logger instance :type _log_level: str, None :returns: The values that will be used for the configuration settings """ _log_levels = { 'general': _log_level, 'file': _file_level, 'console': _console_level, 'syslog': _syslog_level, } _logger_name = LOGGING_DEFAULTS.get('logger_name') if not _logger_name else _logger_name if _debug: for _log_type in _log_levels: _log_levels[_log_type] = 'debug' else: if _log_level: for _lvl_type, _lvl_value in _log_levels.items(): if _lvl_type != 'general' and _lvl_value is None: _log_levels[_lvl_type] = _log_level else: _log_level = LOGGING_DEFAULTS.get('log_level') if _formatter and isinstance(_formatter, str): _formatter = logging.Formatter(_formatter) _formatter = LOGGING_DEFAULTS.get('formatter') if not _formatter else _formatter return _logger_name, _log_levels, _formatter def _get_log_levels_from_dict(_log_levels): """This function returns the individual log level values from a dictionary. :param _log_levels: Dictionary containing log levels for different handlers :type _log_levels: dict :returns: Individual string values for each handler """ _general = _log_levels.get('general') _file = _log_levels.get('file') _console = _log_levels.get('console') _syslog = _log_levels.get('syslog') return _general, _file, _console, _syslog def _set_logging_level(_logger, _log_level): """This function sets the logging level for a :py:class:`logging.Logger` instance. :param _logger: The :py:class:`logging.Logger` instance :type _logger: Logger :param _log_level: The log level as a string (``debug``, ``info``, ``warning``, ``error`` or ``critical``) :type _log_level: str :returns: The :py:class:`logging.Logger` instance with a logging level set where applicable """ if _log_level == 'debug': _logger.setLevel(logging.DEBUG) elif _log_level == 'info': _logger.setLevel(logging.INFO) elif _log_level == 'warning': _logger.setLevel(logging.WARNING) elif _log_level == 'error': _logger.setLevel(logging.ERROR) elif _log_level == 'critical': _logger.setLevel(logging.CRITICAL) return _logger def _add_handlers(_logger, _formatter, _no_output, _file_output, _file_log_level, _log_file, _overwrite_log_files, _console_output, _console_log_level, _syslog_output, _syslog_log_level, _syslog_address, _syslog_port): # TODO: Add docstring if _no_output or not any((_file_output, _console_output, _syslog_output)): _logger.addHandler(logging.NullHandler()) else: if _file_output: # Add the FileHandler to the Logger object _logger = _add_file_handler(_logger, _file_log_level, _log_file, _overwrite_log_files, _formatter) if _console_output: # Add the StreamHandler to the Logger object _logger = _add_stream_handler(_logger, _console_log_level, _formatter) if _syslog_output: # Add the SyslogHandler to the Logger object _logger = _add_syslog_handler(_logger, _syslog_log_level, _formatter, _syslog_address, _syslog_port) return _logger def _add_file_handler(_logger, _log_level, _log_file, _overwrite, _formatter): """This function adds a :py:class:`logging.FileHandler` to the :py:class:`logging.Logger` instance. :param _logger: The :py:class:`logging.Logger` instance :type _logger: Logger :param _log_level: The log level to set for the handler :type _log_level: str :param _log_file: The log file (as a file name or a file path) to which messages should be written .. note:: If a file path isn't provided then the default directory is the home directory of the user instantiating the :py:class:`logging.Logger` object. If a file name is also no provided then it will default to using ``highspot.log`` as the file name. :param _overwrite: Determines if messages should be appended to the file (default) or overwrite it :type _overwrite: bool :param _formatter: The :py:class:`logging.Formatter` to apply to messages passed through the handler :type _formatter: Formatter :returns: The :py:class:`logging.Logger` instance with the added :py:class:`logging.FileHandler` """ # Define the log file to use _home_dir = str(Path.home()) if _log_file: if not any((('/' in _log_file), ('\\' in _log_file))): _log_file = os.path.join(_home_dir, _log_file) else: _log_file = os.path.join(_home_dir, 'highspot.log') # Identify if log file should be overwritten _write_mode = 'w' if _overwrite else 'a' # Instantiate the handler _handler = logging.FileHandler(_log_file, _write_mode) _log_level = HANDLER_DEFAULTS.get('file_log_level') if not _log_level else _log_level _handler = _set_logging_level(_handler, _log_level) _handler.setFormatter(_formatter) # Add the handler to the logger _logger.addHandler(_handler) return _logger def _add_stream_handler(_logger, _log_level, _formatter): """This function adds a :py:class:`logging.StreamHandler` to the :py:class:`logging.Logger` instance. :param _logger: The :py:class:`logging.Logger` instance :type _logger: Logger :param _log_level: The log level to set for the handler :type _log_level: str :param _formatter: The :py:class:`logging.Formatter` to apply to messages passed through the handler :type _formatter: Formatter :returns: The :py:class:`logging.Logger` instance with the added :py:class:`logging.StreamHandler` """ _log_level = HANDLER_DEFAULTS.get('console_log_level') if not _log_level else _log_level _stdout_levels = ['DEBUG', 'INFO'] if _log_level.upper() in _stdout_levels: _logger = _add_split_stream_handlers(_logger, _log_level, _formatter) else: _handler = logging.StreamHandler() _handler = _set_logging_level(_handler, _log_level) _handler.setFormatter(_formatter) _logger.addHandler(_handler) return _logger def _add_split_stream_handlers(_logger, _log_level, _formatter): """This function splits messages into q ``stdout`` or ``stderr`` handler depending on the log level. .. seealso:: Refer to the documentation for the :py:class:`highspot.utils.log_utils.LessThanFilter` for more information on how this filtering is implemented and for credit to the original author. :param _logger: The :py:class:`logging.Logger` instance :type _logger: Logger :param _log_level: The log level provided for the stream handler (i.e. console output) :type _log_level: str :param _formatter: The :py:class:`logging.Formatter` to apply to messages passed through the handlers :type _formatter: Formatter :returns: The logger instance with the two handlers added """ # Configure and add the STDOUT handler _stdout_handler = logging.StreamHandler(sys.stdout) _stdout_handler = _set_logging_level(_stdout_handler, _log_level) _stdout_handler.addFilter(LessThanFilter(logging.WARNING)) _stdout_handler.setFormatter(_formatter) _logger.addHandler(_stdout_handler) # Configure and add the STDERR handler _stderr_handler = logging.StreamHandler(sys.stderr) _stderr_handler.setLevel(logging.WARNING) _stderr_handler.setFormatter(_formatter) _logger.addHandler(_stderr_handler) # Return the logger with the added handlers return _logger def _add_syslog_handler(_logger, _log_level, _formatter, _address, _port): # TODO: Add docstring _log_level = HANDLER_DEFAULTS.get('syslog_log_level') if not _log_level else _log_level _address = HANDLER_DEFAULTS.get('syslog_address') if not _address else _address _port = HANDLER_DEFAULTS.get('syslog_port') if not _port else _port _handler = logging.handlers.SysLogHandler(address=(_address, _port)) _handler = _set_logging_level(_handler, _log_level) _handler.setFormatter(_formatter) _logger.addHandler(_handler) return _logger