ansible-later/ansiblelater/logger.py
2024-01-15 15:26:56 +01:00

211 lines
5.5 KiB
Python

"""Global logging helpers."""
import logging
import os
import sys
import colorama
from pythonjsonlogger import jsonlogger
CONSOLE_FORMAT = "{}%(levelname)s:{} %(message)s"
JSON_FORMAT = "%(asctime)s %(levelname)s %(message)s"
def strtobool(value):
"""Convert a string representation of truth to true or false."""
_map = {
"y": True,
"yes": True,
"t": True,
"true": True,
"on": True,
"1": True,
"n": False,
"no": False,
"f": False,
"false": False,
"off": False,
"0": False,
}
try:
return _map[str(value).lower()]
except KeyError as err:
raise ValueError(f'"{value}" is not a valid bool value') from err
def to_bool(string):
return bool(strtobool(str(string)))
def _should_do_markup():
py_colors = os.environ.get("PY_COLORS", None)
if py_colors is not None:
return to_bool(py_colors)
return sys.stdout.isatty() and os.environ.get("TERM") != "dumb"
colorama.init(autoreset=True, strip=(not _should_do_markup()))
def flag_extra(extra):
"""Ensure extra args are prefixed."""
flagged = {}
if isinstance(extra, dict):
for key, value in extra.items():
flagged["later_" + key] = value
return flagged
class LogFilter:
"""A custom log filter which excludes log messages above the logged level."""
def __init__(self, level):
"""
Initialize a new custom log filter.
:param level: Log level limit
:returns: None
"""
self.__level = level
def filter(self, logRecord): # noqa
# https://docs.python.org/3/library/logging.html#logrecord-attributes
return logRecord.levelno <= self.__level
class MultilineFormatter(logging.Formatter):
"""Logging Formatter to reset color after newline characters."""
def format(self, record):
record.msg = record.msg.replace("\n", f"\n{colorama.Style.RESET_ALL}... ")
record.msg = record.msg + "\n"
return logging.Formatter.format(self, record)
class MultilineJsonFormatter(jsonlogger.JsonFormatter):
"""Logging Formatter to remove newline characters."""
def format(self, record):
record.msg = record.msg.replace("\n", " ")
return jsonlogger.JsonFormatter.format(self, record)
def get_logger(name=None, level=logging.DEBUG, json=False):
"""
Build a logger with the given name and returns the logger.
:param name: The name for the logger. This is usually the module name, `__name__`.
:param level: Initialize the new logger with given log level.
:param json: Boolean flag to enable json formatted log output.
:return: logger object
"""
logger = logging.getLogger(name)
logger.setLevel(level)
logger.addHandler(_get_error_handler(json=json))
logger.addHandler(_get_warn_handler(json=json))
logger.addHandler(_get_info_handler(json=json))
logger.addHandler(_get_critical_handler(json=json))
logger.propagate = False
return logger
def update_logger(logger, level=None, json=None):
"""Update logger configuration to change logging settings."""
for handler in logger.handlers[:]:
logger.removeHandler(handler)
logger.setLevel(level)
logger.addHandler(_get_error_handler(json=json))
logger.addHandler(_get_warn_handler(json=json))
logger.addHandler(_get_info_handler(json=json))
logger.addHandler(_get_critical_handler(json=json))
def _get_error_handler(json=False):
handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.ERROR)
handler.addFilter(LogFilter(logging.ERROR))
handler.setFormatter(MultilineFormatter(error(CONSOLE_FORMAT)))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_warn_handler(json=False):
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.WARN)
handler.addFilter(LogFilter(logging.WARN))
handler.setFormatter(MultilineFormatter(warn(CONSOLE_FORMAT)))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_info_handler(json=False):
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.INFO)
handler.addFilter(LogFilter(logging.INFO))
handler.setFormatter(MultilineFormatter(info(CONSOLE_FORMAT)))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_critical_handler(json=False):
handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.CRITICAL)
handler.addFilter(LogFilter(logging.CRITICAL))
handler.setFormatter(MultilineFormatter(critical(CONSOLE_FORMAT)))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def critical(message):
"""Format critical messages and return string."""
return color_text(colorama.Fore.RED, message)
def error(message):
"""Format error messages and return string."""
return color_text(colorama.Fore.RED, message)
def warn(message):
"""Format warn messages and return string."""
return color_text(colorama.Fore.YELLOW, message)
def info(message):
"""Format info messages and return string."""
return color_text(colorama.Fore.BLUE, message)
def color_text(color, msg):
"""
Colorize strings.
:param color: colorama color settings
:param msg: string to colorize
:returns: string
"""
msg = msg.format(colorama.Style.BRIGHT, colorama.Style.NORMAL)
return f"{color}{msg}{colorama.Style.RESET_ALL}"