1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / rads /
Filename : _logging.py
back
"""Rads logging functions"""

from typing import Literal, Union, IO
import sys
import os
from pathlib import Path
import logging
from logging.handlers import WatchedFileHandler
from .color import red, yellow


def setup_logging(
    path: Union[Path, str, None],
    name: Union[str, None] = None,
    fmt: str = '%(asctime)s %(levelname)s %(message)s',
    datefmt: str = r'%Y-%m-%d %H:%M:%S',
    multiline: bool = True,
    loglevel: Union[int, str] = logging.DEBUG,
    print_out: Union[IO, Literal['stdout', 'stderr'], None] = None,
    print_loglevel: Union[int, str, None] = None,
    chown: Union[tuple[int, int], None] = None,
    chmod: Union[int, None] = None,
) -> logging.Logger:
    """Sets up and returns the root logger, or a named logger if ``name`` is set

    Args:
        path: file path to log to. If set to None, print_out must not be None.
        name: logger name for logging.getLogger()
        fmt: format for ``logging.Formatter``
        datefmt: date format for ``logging.Formatter``
        multiline: whether to support multiline logging
        loglevel: loglevel for logging.setLevel - will accept a constant from
            the logging module such as logging.INFO, or the string of the level
            name, e.g. 'INFO'
        print_out: set this to ``sys.stdout`` or ``sys.stderr`` to
            also print there. Also accepts 'stdout' or 'stderr' as literal str
        print_loglevel: optional separate log level for the ``print_out`` kwarg.
            If unset, it will use the ``loglevel`` kwarg.
        chown: ensure ownership of the log path
            Only valid if path is set.
        chmod: ensure perms of the log path *in octal*.
            Only valid if path is set.
    """
    if isinstance(loglevel, str):
        loglevel = getattr(logging, loglevel.upper())
        assert isinstance(loglevel, int)
    if isinstance(print_loglevel, str):
        print_loglevel = getattr(logging, print_loglevel.upper())
        assert isinstance(print_loglevel, int)
    if isinstance(print_out, str):
        if print_out.lower() == 'stdout':
            print_out = sys.stdout
        elif print_out.lower() == 'stderr':
            print_out = sys.stderr
        else:
            raise TypeError(print_out)
    if path:
        path = Path(path)
        path.touch(mode=0o644 if chmod is None else chmod, exist_ok=True)
    elif not print_out:
        raise TypeError("At least one of 'path' and/or 'print_out' must be set")
    if chmod is not None:
        if not path:
            raise TypeError("'path' must be set to use 'chmod'")
        os.chmod(path, chmod)
    if chown is not None:
        if not isinstance(chown, tuple):
            raise TypeError("'chown' must be a tuple")
        if not path:
            raise TypeError("'path' must be set to use 'chown'")
        os.chown(path, chown[0], chown[1])
    logger = logging.getLogger(name)
    if multiline:
        formatter = MultilineFormatter(fmt=fmt, datefmt=datefmt)
    else:
        formatter = logging.Formatter(fmt=fmt, datefmt=datefmt)
    if path:
        main_handler = WatchedFileHandler(path)
        main_handler.setFormatter(formatter)
        main_handler.setLevel(loglevel)
        logger.addHandler(main_handler)
    if print_out:
        print_handler = logging.StreamHandler(stream=print_out)
        print_handler.setFormatter(formatter)
        print_handler.setLevel(print_loglevel or loglevel)
        logger.addHandler(print_handler)
    levels = [loglevel]
    if print_loglevel is not None:
        levels.append(print_loglevel)
    logger.setLevel(min(levels))
    return logger


setup_logging.__module__ = 'rads'


def setup_verbosity(
    loglevel: Union[int, str] = 'DEBUG',
    color: Union[bool, None] = None,
    name: Union[str, None] = 'rads_verbosity',
):
    """Return a custom logger used to easily handle error and message printing.
        debug & info: prints to stdout
        warning: prints to stderr (in yellow, if enabled)
        error & critical: prints to stderr (in red, if enabled)

    Args:
        loglevel: filter to only print up to this level.
            This is especially useful to add --verbose/--quiet behavior
        color: set this True or False to force colors on or off. If unset,
            it will check if stderr is a TTY and enable colors if so
        name: name of the logger for logging.getLogger()

    Returns:
        Logger: the configured Logger object
    """
    if isinstance(loglevel, str):
        loglevel = getattr(logging, loglevel.upper())
        assert isinstance(loglevel, int)
    logger = logging.getLogger(name)
    stdout_handler = logging.StreamHandler(stream=sys.stdout)
    stdout_handler.setFormatter(logging.Formatter(fmt='%(message)s'))
    stdout_handler.addFilter(LevelFilter(logging.DEBUG, logging.INFO))
    if color is None:
        color = hasattr(sys.stderr, 'isatty') and sys.stdout.isatty()
    if color:
        err_fmt = logging.Formatter(fmt=red('%(message)s'))
        warn_fmt = logging.Formatter(fmt=yellow('%(message)s'))
    else:
        warn_fmt = err_fmt = logging.Formatter(fmt='%(message)s')
    warning_handler = logging.StreamHandler(stream=sys.stderr)
    warning_handler.setFormatter(warn_fmt)
    warning_handler.addFilter(LevelFilter(logging.WARNING, logging.WARNING))
    error_handler = logging.StreamHandler(stream=sys.stderr)
    error_handler.setFormatter(err_fmt)
    error_handler.addFilter(LevelFilter(logging.ERROR, logging.CRITICAL))
    logger.addHandler(stdout_handler)
    logger.addHandler(warning_handler)
    logger.addHandler(error_handler)
    logger.setLevel(loglevel)
    return logger


setup_verbosity.__module__ = 'rads'


class LevelFilter(logging.Filter):
    """Allows setting both a min and max log level via log.addFilter instead of
    log.setLevel"""

    __module__ = 'rads'

    def __init__(self, low, high):
        self._low = low
        self._high = high
        logging.Filter.__init__(self)

    def filter(self, record):
        if self._low <= record.levelno <= self._high:
            return True
        return False


class MultilineFormatter(logging.Formatter):
    """Subclass of logging.Formatter that can handle multiline strings.
    rads.setup_logging() will use this by default unless multiline=False"""

    __module__ = 'rads'

    def format(self, record: logging.LogRecord):
        save_msg = f'{record.msg}'
        output = ""
        for index, line in enumerate(save_msg.splitlines()):
            record.msg = line
            if index > 0:
                output += "\n"
            output += super().format(record)
        record.msg = save_msg
        record.message = output
        return output