Source code for gemseo.utils.logging_tools

# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
# Contributors:
#    INITIAL AUTHORS - initial API and implementation and/or initial
#                           documentation
#        :author: Antoine Dechaume
#    OTHER AUTHORS   - MACROSCOPIC CHANGES
"""Logging tools."""

from __future__ import annotations

import types
from dataclasses import dataclass
from logging import WARNING
from logging import FileHandler
from logging import Formatter
from logging import Logger
from logging import LogRecord
from logging import StreamHandler
from logging import getLogger
from sys import stderr
from sys import stdout
from types import TracebackType
from typing import Final

DEFAULT_DATE_FORMAT: Final[str] = "%H:%M:%S"
"""The format of the date of the logged message."""

DEFAULT_MESSAGE_FORMAT: Final[str] = "%(levelname)8s - %(asctime)s: %(message)s"
"""The format of the logged message."""


GEMSEO_LOGGER: Final[Logger] = getLogger("gemseo")
"""The GEMSEO's logger."""


[docs] @dataclass class LoggingSettings: """The settings of a logger.""" date_format: str = DEFAULT_DATE_FORMAT """The format of the date of the logged message.""" message_format: str = DEFAULT_MESSAGE_FORMAT """The format of the logged message.""" logger: Logger = GEMSEO_LOGGER """The logger."""
LOGGING_SETTINGS = LoggingSettings() """The logging settings. The parameters are changed by :func:`.configure_logger`. """
[docs] class MultiLineHandlerMixin: """Stateless mixin class to override logging handlers behavior.""" @staticmethod def __get_raw_record_message(record): """Return the raw message of a log record.""" return record.msg
[docs] def emit(self, record) -> None: """Emit one logging message per input record line.""" # compute the message without the logging prefixes (timestamp, level, ...) message = record.getMessage() # replace getMessage so the message is not computed again when emitting # each line record.getMessage = types.MethodType(self.__get_raw_record_message, record) # backup old raw message old_msg = record.msg for line in message.split("\n"): record.msg = line super().emit(record) # restore genuine getMessage and raw message for other handlers record.getMessage = types.MethodType(LogRecord.getMessage, record) record.msg = old_msg
[docs] class MultiLineStreamHandler(MultiLineHandlerMixin, StreamHandler): """StreamHandler to split multiline logging messages."""
[docs] class MultiLineFileHandler(MultiLineHandlerMixin, FileHandler): """FileHandler to split multiline logging messages."""
[docs] class LoggingContext: """Context manager for selective logging. Change the level of the logger in a ``with`` block. Examples: >>> import logging >>> logger = logging.getLogger() >>> logger.setLevel(logging.INFO) >>> logger.info("This should appear.") >>> with LoggingContext(logger): >>> logger.warning("This should appear.") >>> logger.info("This should not appear.") >>> >>> logger.info("This should appear.") Source: `Logging Cookbook <https://docs.python.org/3/howto/ logging-cookbook.html#using-a-context-manager-for-selective-logging>`_ """ logger: Logger """The logger.""" level: int | None """The level of the logger to be used on block entry. If ``None``, do not change the level of the logger. """ handler: StreamHandler """An additional handler to be used on block entry.""" close: bool """Whether to close the handler on block exit.""" def __init__( self, logger: Logger, level: int | None = WARNING, handler: StreamHandler | None = None, close: bool = True, ) -> None: """ Args: logger: The logger. level: The level of the logger to be used on block entry. If ``None``, do not change the level of the logger. handler: An additional handler to be used on block entry. close: Whether to close the handler on block exit. """ # noqa:D205 D212 D415 self.logger = logger self.level = level self.handler = handler self.close = close def __enter__(self) -> None: if self.level is not None: self.old_level = self.logger.level self.logger.setLevel(self.level) if self.handler: self.logger.addHandler(self.handler) def __exit__( self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: if self.level is not None: self.logger.setLevel(self.old_level) if self.handler: self.logger.removeHandler(self.handler) if self.handler and self.close: self.handler.close()
# implicit return of None => don't swallow exceptions
[docs] class OneLineLogging: """A context manager to make the StreamHandlers use only one line. Each record replaces the previous one. """ __handler: StreamHandler | None """The handler used by the context manager if any.""" __terminators: tuple[str] """One line terminator per stream handler.""" __formatters: tuple[Formatter] """One formatter per stream handler.""" __logger: Logger """The logger used by the context manager when the handler is not ``None``.""" def __init__(self, logger: Logger) -> None: """ Args: logger: The logger. """ # noqa: D205 D212 D415 self.__handler = None self.__formatter = () self.__terminator = () while logger: self.__logger = logger stop = False for handler in logger.handlers: if isinstance(handler, StreamHandler) and handler.stream in { stdout, stderr, }: self.__handler = handler self.__terminator = handler.terminator self.__formatter = handler.formatter stop = True if stop: break if not logger.propagate: break logger = logger.parent def __enter__(self) -> None: if self.__handler is not None: self.__handler.terminator = "" self.__handler.formatter = Formatter( fmt="\r" + LOGGING_SETTINGS.message_format, datefmt=LOGGING_SETTINGS.date_format, ) def __exit__( self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: if self.__handler is not None: self.__handler.terminator = self.__terminator self.__handler.formatter = self.__formatter self.__handler.stream.write("\n")