Source code for kiwi.logger

# Copyright (c) 2015 SUSE Linux GmbH.  All rights reserved.
# This file is part of kiwi.
# kiwi is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# kiwi is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with kiwi.  If not, see <>
from typing import (
    Dict, Optional, List
import logging
from kiwi.logger_socket import PlainTextSocketHandler
import sys

# project
from kiwi.logger_color_formatter import ColorFormatter
from kiwi.logger_filter import (

from kiwi.exceptions import (

[docs] class Logger(logging.Logger): """ **Extended logging facility based on Python logging** :param str name: name of the logger """ def __init__(self, name: str): logging.Logger.__init__(self, name) self.log_handlers: Dict = {} self.logfile: Optional[str] = None # log INFO to stdout self._add_stream_handler( 'info', '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', [InfoFilter(), LoggerSchedulerFilter()] ) # log WARNING messages to stdout self._add_stream_handler( 'warning', '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', [WarningFilter()] ) # log DEBUG messages to stdout self._add_stream_handler( 'debug', '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', [DebugFilter()] ) # log ERROR messages to stderr self._add_stream_handler( 'error', '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', [ErrorFilter()], sys.__stderr__ ) self.log_level = self.level self.log_flags: Dict[str, bool] = {}
[docs] def getLogLevel(self) -> int: """ Return currently used log level :return: log level number :rtype: int """ return self.log_level
[docs] def getLogFlags(self) -> Dict[str, bool]: """ Return logging flags :return: Dictionary with flags and their activation status :rtype: dict """ return self.log_flags
[docs] def setLogLevel(self, level: int, except_for: List[str] = []) -> None: """ Set custom log level for all console handlers :param int level: log level number """ self.log_level = level for handler_type in self.log_handlers: if handler_type not in except_for: self.log_handlers[handler_type].setLevel(level)
[docs] def setLogFlag(self, flag: str, value: bool = True) -> None: """ Set logging flag for further properties of the logging facility Available flags are: * run-scripts-in-screen :param str flag: name """ self.log_flags[flag] = value
[docs] def set_color_format(self) -> None: """ Set color format for all console handlers """ for handler_type in self.log_handlers: message_format = None if handler_type == 'debug': message_format = \ '$LIGHTCOLOR[ %(levelname)-8s]: %(asctime)-8s | %(message)s' elif handler_type == 'warning' or handler_type == 'error': message_format = \ '$COLOR[ %(levelname)-8s]: %(asctime)-8s | %(message)s' if message_format: self.log_handlers[handler_type].setFormatter( ColorFormatter(message_format, '%H:%M:%S') )
[docs] def set_logfile(self, filename: str) -> None: """ Set logfile handler :param str filename: logfile file path """ try: if filename == 'stdout': # special case, log usual log file contents to stdout handler = logging.StreamHandler(sys.__stdout__) else: handler = logging.FileHandler( filename=filename, encoding='utf-8' ) self.logfile = filename handler.setFormatter( logging.Formatter( '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', '%H:%M:%S' ) ) handler.addFilter(LoggerSchedulerFilter()) self.addHandler(handler) self.log_handlers['file'] = handler except Exception as e: raise KiwiLogFileSetupFailed( '%s: %s' % (type(e).__name__, format(e)) )
[docs] def set_log_socket(self, filename: str) -> None: """ Set log socket handler :param str filename: UDS socket file path. Note if there is no server listening on the socket the log handler setup will fail """ try: handler = PlainTextSocketHandler(filename, None) handler.makeSocket() handler.setFormatter( logging.Formatter( '[ %(levelname)-8s]: %(asctime)-8s | %(message)s', '%H:%M:%S' ) ) handler.addFilter(LoggerSchedulerFilter()) self.addHandler(handler) self.log_handlers['socket'] = handler except Exception as e: raise KiwiLogSocketSetupFailed( 'UDS socket: {0}:{1}: {2}'.format(filename, type(e).__name__, e) )
[docs] def get_logfile(self) -> Optional[str]: """ Return file path name of logfile :return: file path :rtype: str """ return self.logfile
[docs] @staticmethod def progress( current: int, total: int, prefix: str, bar_length: int = 40 ) -> None: """ Custom progress log information. progress information is intentionally only logged to stdout and will bypass any handlers. We don't want this information to show up in the log file :param int current: current item :param int total: total number of items :param string prefix: prefix name :param int bar_length: length of progress bar """ try: percent = float(current) / total except Exception: # we don't want the progress to raise an exception # In case of any error e.g division by zero the current # way out is to skip the progress update return hashes = '#' * int(round(percent * bar_length)) spaces = ' ' * (bar_length - len(hashes)) sys.stdout.write('\r{0}: [{1}] {2}%'.format( prefix, hashes + spaces, int(round(percent * 100)) )) if current == 100: sys.stdout.write('\n') sys.stdout.flush()
def _add_stream_handler( self, handler_type, message_format, message_filter, channel=sys.__stdout__ ): handler = logging.StreamHandler(channel) handler.setFormatter( logging.Formatter(message_format, '%H:%M:%S') ) for rule in message_filter: handler.addFilter(rule) self.addHandler(handler) self.log_handlers[handler_type] = handler