"""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]@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
# third party logger name pattern white list
THIRD_PARTY_LOGGER_WHITE_LIST = ["sdkloop.*", "luna3.*"]
[docs]def checkThirdPartyLoggerName(name, thirdPartyLoggerWhitList: list[str]) -> bool:
"""
Check third party logger name that is allowed.
Args:
name: third party logger name
thirdPartyLoggerWhitList: third party logger whit list. allowed format:
- strict name
- wildcard `rootname.*`
Returns:
true if name matches with any name from thirdPartyLoggerWhitList otherwise false
>>> checkThirdPartyLoggerName("sdkloop.facedetector", THIRD_PARTY_LOGGER_WHITE_LIST)
True
>>> checkThirdPartyLoggerName("aiohttp", THIRD_PARTY_LOGGER_WHITE_LIST)
False
"""
for allowedLoggerName in thirdPartyLoggerWhitList:
if allowedLoggerName.endswith(".*"):
if name.startswith(allowedLoggerName[:-2]):
return True
elif allowedLoggerName == name:
return True
return False
[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
if not checkThirdPartyLoggerName(thirdPartyLogger.name, THIRD_PARTY_LOGGER_WHITE_LIST):
thirdPartyLogger.level = max(logging.WARNING, cls._settings.logLevel)
else:
thirdPartyLogger.level = cls._settings.logLevel
# 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
