Source code for luna_python_matcher.crutches_on_wheels.utils.log

"""Loggers

Loggers module.
"""
import logging
import sys
import time
from contextlib import contextmanager
from dataclasses import dataclass
from enum import Enum
from logging import Logger as BaseLogger, _acquireLock, _releaseLock
from logging.handlers import RotatingFileHandler
from os import path
from typing import Optional, Union, List, Any, final

from .rid import DEFAULT_REQUEST_ID, requestIdCtx

DEBUG_DEFAULT_FORMAT_STRING = (
    "[{process:07d} {asctime}] {levelname} [{filename} {lineno}]: {appName} {requestId}{channel}: {message}"
)

# (str): format string for non debug logs
NOT_DEBUG_DEFAULT_FORMAT_STRING = "[{process:07d} {asctime}] {levelname}: {appName} {requestId}{channel}: {message}"


# (int): size of one megabyte in bytes
SIZE_OF_MB = 2 ** 20


def _adaptChannelForRecordFormat(channel: str) -> str:
    """Adopt channel name for record"""
    return f": {channel}" if channel else ""


[docs]def generateStdoutLogHandler() -> logging.StreamHandler: """ Generate stout log handler Returns: new stdout handler """ handler = logging.StreamHandler(sys.stdout) return handler
[docs]def setUpHandler(handler: logging.Handler, multilineStackTrace: bool, level: int): """ Setup log handler. set log record formatter, level Args: handler: handler multilineStackTrace: allow a multiline line stack trace message level: log level """ handler.level = level if level == logging.DEBUG: fmtString = DEBUG_DEFAULT_FORMAT_STRING else: fmtString = NOT_DEBUG_DEFAULT_FORMAT_STRING if multilineStackTrace: handler.setFormatter(Formatter(fmt=fmtString, style="{")) else: handler.setFormatter(OneLineExceptionFormatter(fmt=fmtString, style="{"))
[docs]def generateFileHandler(logFileName: str, maxSize: int = 1024) -> logging.FileHandler: """ Generate file handler Args: logFileName: filename maxSize: max size of log file Returns: rotating file handler if maxSize> 0 otherwise usual file handler """ if maxSize > 0: handler = RotatingFileHandler(logFileName, maxBytes=maxSize * SIZE_OF_MB, backupCount=5,) else: handler = logging.FileHandler(logFileName) return handler
[docs]@contextmanager def logHandlerLock(): """ Contextmanager for access management to logger handlers. All changes handlers operations must be thread safe. """ _acquireLock() try: yield finally: _releaseLock()
[docs]class Formatter(logging.Formatter): """ Log record formater """ # time converter converter = time.localtime # default milliseconds format default_msec_format = "%s.%03d000"
[docs]class OneLineExceptionFormatter(Formatter): """ One line exception log record formatter References: https://docs.python.org/3/howto/logging-cookbook.html#customized-exception-formatting """
[docs] def format(self, record: logging.LogRecord) -> str: """ Format an exception so that it prints on a single line. """ msg = super().format(record) if record.exc_text: msg = msg.replace("\n", " |") return msg
[docs]@dataclass class LogSettings: """ Logger settings container """ # app name appName: str # log level logLevel: int # log records time logTime: str # folder for logs folderForLog: str # max size of log file maxSize: int # log file suffix logNameSuffix: Union[str, None] # write or not logs to stdout logToStdOut: bool # write or not logs to file logToFile: bool # allow a multiline line stack trace message multilineStackTrace: bool
[docs]class LogDest(Enum): """ Log destination """ file = "file" #: destination is log file stdout = "stdout" #: destination is stdout
[docs]class Logger(BaseLogger): """ Application logger. Global logger class. This class realize following features: 1. Class contains global handlers. All loggers will be to have this handlers. 2. Class replaces itself default logger class from standard library (logging.Logger). All third party loggers (based on logging.Logger) will be use handlers of this class 3. method reinitialize updates all created loggers. All features based on class variable `_handlers` and property `handlers`. They patch attribute `handlers` of logger from standard library and make them globals. """ # logger settings _settings: Optional[LogSettings] = None # logs fie handlers _logsFileHandlers: List[logging.FileHandler] = [] # stdout handler _stdoutHandler: Optional[logging.StreamHandler] = None # logger default request id defaultRequestId: str = DEFAULT_REQUEST_ID # global logger handlers _handlers = []
[docs] @staticmethod def getLogLevel(logLevel) -> int: """ Get log level from config for logger. Returns: int: if logLevel not set or incorrect will return logging.NOTSET """ if logLevel == "DEBUG": return logging.DEBUG if logLevel == "ERROR": return logging.ERROR if logLevel == "INFO": return logging.INFO if logLevel == "WARNING": return logging.WARNING return logging.NOTSET
[docs] @classmethod def initiate( cls, appName: str = "", logLevel: str = "DEBUG", logTime: str = "LOCAL", folderForLog: str = "./", maxSize: int = 1024, logNameSuffix: Optional[str] = None, logToStdOut: bool = True, logToFile: bool = True, multilineStackTrace: bool = True, ) -> None: """ Initiate class settings. Args: appName: application name logLevel: log level logTime: time of logs folderForLog: folder with log-files maxSize: max log file size, 1024 mb for default, 0 - disable rotate logNameSuffix: optional suffix name for log name files. logToStdOut: flush log to stdout logToFile: flush log to a log file multilineStackTrace: allow a multiline line stack trace message """ cls.reInitialize( appName=appName, logToFile=logToFile, logToStdOut=logToStdOut, multilineStackTrace=multilineStackTrace, maxSize=maxSize, logNameSuffix=logNameSuffix, folderForLog=folderForLog, logLevel=logLevel, logTime=logTime, ) cls.updateDefaultPythonLoggers()
@classmethod def _update3partyLogger(cls, thirdPartyLogger: logging.Logger): """ Update 3-party logger. set up handlers, log level and other Args: thirdPartyLogger: 3-party logger """ if isinstance(thirdPartyLogger, cls): return thirdPartyLogger.level = logging.WARNING # close old handlers for handler in thirdPartyLogger.handlers: handler.close() # setup global hendlers with logHandlerLock(): thirdPartyLogger.handlers = cls._handlers # set up a record updating with extra data def _filter(record): record.appName = cls._settings.appName record.requestId = cls.defaultRequestId record.channel = _adaptChannelForRecordFormat(thirdPartyLogger.name) return thirdPartyLogger.__class__.filter(thirdPartyLogger, record) thirdPartyLogger.filter = _filter
[docs] @classmethod def updateDefaultPythonLoggers(cls): """ setup this class as default class for 3-party loggers, which use logging.getLogger, manager.getLogger, getLoggerClass. """ # upgrade already created loggers with logHandlerLock(): for thirdPartylogger in logging.Logger.manager.loggerDict.values(): # set up created 3-party loggers if not isinstance(thirdPartylogger, BaseLogger): continue if isinstance(thirdPartylogger, Logger): # skip our loggers continue cls._update3partyLogger(thirdPartylogger) class ThirdPartyLoggers(Logger): """Third party logger factory""" level = logging.WARNING # set up class for future loggers cls.manager.setLoggerClass(ThirdPartyLoggers) logging.setLoggerClass(ThirdPartyLoggers)
[docs] @classmethod def reInitializeStdoutLogging(cls, settings: LogSettings): """ Re-Initialize stdout log handlers. Closes old handlers and create new if it iis needed Args: settings: settings for re-initialization """ if cls._stdoutHandler: cls.removeHandler(cls._stdoutHandler) if not cls._stdoutHandler.stream.closed: # Let capsys monkeypatch the stream cls._stdoutHandler.close() cls._stdoutHandler = None if settings.logToStdOut: cls._stdoutHandler = generateStdoutLogHandler() setUpHandler(cls._stdoutHandler, multilineStackTrace=settings.multilineStackTrace, level=settings.logLevel) cls.addHandler(cls._stdoutHandler)
[docs] @classmethod @final def addHandler(cls, hdlr: logging.Handler) -> None: """ Add handlers to all loggers Overwrite original method. Before it was instance method, now - global classmethod Args: hdlr: log handler """ with logHandlerLock(): if hdlr not in cls._handlers: cls._handlers.append(hdlr)
[docs] @classmethod @final def removeHandler(cls, hdlr: logging.Handler): """ Remove the specified handler from all loggers. Overwrite original method. Before it was instance method, now - global classmethod Args: hdlr: log handler """ with logHandlerLock(): if hdlr in cls._handlers: cls._handlers.remove(hdlr)
[docs] @classmethod def reInitializeFileLogging(cls, settings: LogSettings): """ Re-Initialize file logs handlers. Closes old handlers and create new if it iis needed Args: settings: settings for re-initialization """ for handler in cls._logsFileHandlers: handler.close() cls.removeHandler(handler) cls._logsFileHandlers = [] if settings.logToFile: folderForLog = settings.folderForLog appName = settings.appName logNameSuffix = settings.logNameSuffix multilineStackTrace = settings.multilineStackTrace for fileLogLevel in ("INFO", "ERROR", "DEBUG", "WARNING"): if settings.logLevel > getattr(logging, fileLogLevel): continue if not logNameSuffix: logFileName = path.join(folderForLog, f"{appName}_{fileLogLevel}.txt") else: logFileName = path.join(folderForLog, f"{appName}_{fileLogLevel}_{logNameSuffix}.txt") mStackTrace = multilineStackTrace if fileLogLevel != "DEBUG" else True handler = generateFileHandler(logFileName, settings.maxSize) setUpHandler(handler, mStackTrace, level=getattr(logging, fileLogLevel)) cls.addHandler(handler) cls._logsFileHandlers.append(handler)
[docs] @classmethod def reInitializeLogTime(cls, settings: LogSettings): """ Re Initialize log time. Update time converter of global formatter Args: settings: settings for re-initialization """ Formatter.converter = time.localtime if settings.logTime == "LOCAL" else time.gmtime
[docs] @classmethod def reInitialize( cls, appName: str = "", logLevel: str = "DEBUG", logTime: str = "LOCAL", folderForLog: str = "./", maxSize: int = 1024, logNameSuffix: Optional[str] = None, logToStdOut: bool = True, logToFile: bool = True, multilineStackTrace: bool = True, force: bool = False, ) -> None: """ Re-Initiate class settings. Args: appName: application name logLevel: log level logTime: time of logs folderForLog: folder with log-files maxSize: max log file size, 1024 mb for default, 0 - disable rotate logNameSuffix: optional suffix name for log name files. logToStdOut: flush log to stdout logToFile: flush log to a log file multilineStackTrace: allow a multiline line stack trace message force: reinitialize all settings """ _settings = LogSettings( maxSize=maxSize, appName=appName, logTime=logTime, logLevel=cls.getLogLevel(logLevel), logNameSuffix=logNameSuffix, logToFile=logToFile, logToStdOut=logToStdOut, folderForLog=folderForLog, multilineStackTrace=multilineStackTrace, ) if _settings == cls._settings and not force: return try: cls.reInitializeStdoutLogging(_settings) cls.reInitializeFileLogging(_settings) cls.reInitializeLogTime(_settings) except Exception: # fallback to old settings, try set correct Logger state for logging errors if cls._settings is not None: cls.reInitializeStdoutLogging(cls._settings) cls.reInitializeFileLogging(cls._settings) cls.reInitializeLogTime(cls._settings) raise cls._settings = _settings
@property def logTime(self) -> str: """ Get current log time format Returns: logTime from settings """ return self._settings.logTime @property def level(self) -> int: """ Overwrite property. Not save log level to an in instance of the Logger for correct reinitialize loggers. Returns: class log level """ return self._settings.logLevel @level.setter def level(self, level: int): """ Set log level stub. need for supper().__init__ Args: level: new log level Warnings: Use only `initiate` and `reInitialize` functions """ @property def requestId(self) -> str: """ Get current request id Returns: request id. priority: 1) request id from __init__ 2) from requestIdCtx 3) default """ return self._requestId or requestIdCtx.get() or self.defaultRequestId @property @final def handlers(self) -> List[logging.Handler]: """ Get list of global handlers. Rewrite local handlers from logging.Logger Not save handlers to an in instance of the Logger for correct reinitialize loggers. All created loggers have these handlers. Returns: list of global handlers. """ return self._handlers @handlers.setter @final def handlers(self, handlers: Any): """ Set handlers stub. need for supper().__init__ Args: handlers: new handlers Warnings: Use only `initiate` and `reInitialize` functions """
[docs] def filter(self, record: logging.LogRecord) -> bool: """ Add extra data (request id and app name) to record Args: record: record Returns: result of filters """ record.appName = self._settings.appName record.requestId = self.requestId record.channel = _adaptChannelForRecordFormat(self.channel) return super().filter(record)
[docs] def exception(self, *args, **kwargs,) -> None: #: pylint: disable-msg=W0221 """ Exception without msg support. Args: *args: args **kwargs: kwargs """ if len(args) == 0 and "msg" not in kwargs: return super().exception("Uncaught exception occurred", *args, **kwargs) return super().exception(*args, **kwargs)
def __init__(self, template: str = "", requestId: Optional[str] = None,) -> None: # pylint: disable-msg=W0613 """ Init logger. Args: requestId: user defined request id template: string for marking logs. Typical usage - request id. """ self._requestId = requestId self.channel = template super().__init__(template)
# global service logger logger = Logger() #: pylint: disable-msg=C0103