pocketutils.misc.fancy_loguru.InterceptHandler.emit() - Code Metrics - Inspection of "fix: bugs" - dmyersturnbull/pocketutils - Measure and Improve Code Quality continuously with Scrutinizer

Passed
Push — main ( f8490e...e59e89 )

by Douglas
created 2021-10-12 02:22 UTC
InterceptHandler.emit() A

↳ Parent: pocketutils.misc.fancy_loguru
Complexity

Conditions
Size

Total Lines	12
Code Lines	10
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
cc	3
eloc	10
nop	2
dl	0
loc	12
rs	9.9
c	0
b	0
f	0
"""
Loguru logging extension that, configurably:
- redirects built-in logging to your loguru logger
- remembers the handlers added
- auto-detects compression and serialization from filenames
- includes extras in non-serializing handlers
- has a few alternative (possibly better) choices for colors, icons, and levels
- will complain when you do really stupid things
- has a convenient notation for configuring from a CLI
  (e.g. ``--stderr debug --log :INFO:run.log.gz``)
- mandes utf-8
"""
from __future__ import annotations

import abc
import logging
import os
import sys
from dataclasses import dataclass
from inspect import cleandoc
from pathlib import Path
from typing import (
    AbstractSet,
    Any,
    Callable,
    Generic,
    Mapping,
    Optional,
    Sequence,
    TextIO,
    Type,
    TypeVar,
    Union,
)

import loguru._defaults as _defaults


# noinspection PyProtectedMember
import regex

from loguru import logger


# noinspection PyProtectedMember
# noinspection PyProtectedMember
from loguru._logger import Logger


from pocketutils.core.exceptions import IllegalStateError, XValueError

Formatter = Union[str, Callable[[Mapping[str, Any]], str]]
_FMT = cleandoc(
    r"""
    <bold>{time:YYYY-MM-DD HH:mm:ss.SSS}</bold> |
    <level>{level: <8}</level> |
    <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan>
    — <level>{message}[EXTRA]</level>
    {exception}
    """
).replace("\n", " ")


class FormatFactory:

    @classmethod
    def with_extras(

        cls,

        *,

        fmt: str = _FMT,

        sep: str = "; ",

        eq_sign: str = " ",

    ) -> Callable[[Mapping[str, Any]], str]:
        def FMT(record: Mapping[str, Any]) -> str:

            extra = sep.join([e + eq_sign + "{extra[" + e + "]}" for e in record["extra"].keys()])
            if len(extra) > 0:
                extra = f" [ {extra} ]"
            return fmt.replace("[EXTRA]", extra) + os.linesep

        return FMT

    @classmethod
    def plain(cls, *, fmt: str = _FMT) -> Callable[[Mapping[str, Any]], str]:

        def FMT(record: Mapping[str, Any]) -> str:

            return fmt.replace("[EXTRA]", "") + os.linesep

        return FMT


class _SENTINEL:
    pass


T = TypeVar("T", covariant=True, bound=Logger)

Z = TypeVar("Z", covariant=True, bound=Logger)



_LOGGER_ARG_PATTERN = regex.compile(r"(?:([a-zA-Z]+):)?(.*)", flags=regex.V1)
log_compressions = {
    ".xz",
    ".lzma",
    ".gz",
    ".zip",
    ".bz2",
    ".tar",
    ".tar.gz",
    ".tar.bz2",
    ".tar.xz",
}
valid_log_suffixes = {
    *{f".log{c}" for c in log_compressions},
    *{f".txt{c}" for c in log_compressions},
    *{f".json{c}" for c in log_compressions},
}


class FancyLoguruDefaults:

    levels_built_in = dict(
        TRACE=_defaults.LOGURU_TRACE_NO,
        DEBUG=_defaults.LOGURU_DEBUG_NO,
        INFO=_defaults.LOGURU_INFO_NO,
        WARNING=_defaults.LOGURU_WARNING_NO,
        ERROR=_defaults.LOGURU_ERROR_NO,
        CRITICAL=_defaults.LOGURU_CRITICAL_NO,
    )

    # the levels for caution and notice are DEFINED here
    # trace and success must match loguru's
    # and the rest must match logging's
    # note that most of these alternate between informative and problematic
    # i.e. info (ok), caution (bad), success (ok), warning (bad), notice (ok), error (bad)
    levels_extended = {
        **levels_built_in,
        **dict(
            CAUTION=23,
            SUCCESS=25,
            NOTICE=35,
        ),
    }

    colors_built_in = dict(
        TRACE=_defaults.LOGURU_TRACE_COLOR,
        DEBUG=_defaults.LOGURU_DEBUG_COLOR,
        INFO=_defaults.LOGURU_INFO_COLOR,
        SUCCESS=_defaults.LOGURU_SUCCESS_COLOR,
        WARNING=_defaults.LOGURU_WARNING_COLOR,
        ERROR=_defaults.LOGURU_ERROR_COLOR,
        CRITICAL=_defaults.LOGURU_CRITICAL_COLOR,
    )

    colors_extended = dict(
        **colors_built_in,
        CAUTION=_defaults.LOGURU_WARNING_COLOR,
        NOTICE=_defaults.LOGURU_INFO_COLOR,
    )

    colors_red_green_safe = dict(
        TRACE="<dim>",
        DEBUG="<dim>",
        INFO="<bold>",
        CAUTION="<yellow>",
        SUCCESS="<blue>",
        WARNING="<yellow>",
        NOTICE="<blue>",
        ERROR="<red>",
        CRITICAL="<red>",
    )

    icons_built_in = dict(
        TRACE=_defaults.LOGURU_TRACE_ICON,
        DEBUG=_defaults.LOGURU_DEBUG_ICON,
        INFO=_defaults.LOGURU_INFO_ICON,
        SUCCESS=_defaults.LOGURU_SUCCESS_ICON,
        WARNING=_defaults.LOGURU_WARNING_ICON,
        ERROR=_defaults.LOGURU_ERROR_ICON,
        CRITICAL=_defaults.LOGURU_CRITICAL_ICON,
    )

    icons_extended = dict(
        **icons_built_in,
        CAUTION="⚐",
        NOTICE="★",
    )

    level: str = "INFO"

    fmt_simplified = FormatFactory.with_extras()
    fmt_built_in = FormatFactory.plain()
    fmt_built_in_raw = _defaults.LOGURU_FORMAT

    aliases = dict(NONE=None, NO=None, OFF=None, VERBOSE="INFO", QUIET="ERROR")


@dataclass(frozen=True, repr=True, order=True)

class HandlerInfo:
    hid: int
    path: Optional[Path]
    level: Optional[str]
    fmt: Formatter


@dataclass(frozen=False, repr=True)
class _HandlerInfo:
    hid: int
    sink: Any
    level: Optional[int]
    fmt: Formatter


@dataclass(frozen=True, repr=True, order=True)

class LogSinkInfo:
    path: Path
    base: Path
    suffix: str
    serialize: bool
    compression: Optional[str]

    @classmethod
    def guess(cls, path: Union[str, Path]) -> LogSinkInfo:

        path = Path(path)
        base, compression = path.name, None
        for c in log_compressions:

            if path.name.endswith(c):
                base, compression = path.name[: -len(c)], c
        if not [base.endswith(s) for s in [".json", ".log", ".txt"]]:
            raise XValueError(
                f"Log filename {path.name} is not .json, .log, .txt, or a compressed variant"
            )
        return LogSinkInfo(
            path=path,
            base=path.parent / base,
            suffix=compression,
            serialize=base.endswith(".json"),
            compression=compression,
        )


class InterceptHandler(logging.Handler):
    """
    Redirects standard logging to loguru.
    """

    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno
        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


class FancyLoguru(Generic[T]):

    def __init__(self, log: T = logger):
        self._levels = dict(FancyLoguruDefaults.levels_built_in)
        self._logger = log
        self._main: _HandlerInfo = None
        self._paths = {}
        self._aliases = dict(FancyLoguruDefaults.aliases)

    @staticmethod
    def new(t: Type[Z]) -> FancyLoguru[Z]:

        return FancyLoguru[Z](logger)

    @property
    def logger(self) -> T:

        return self._logger

    @property
    def levels(self) -> Mapping[str, int]:

        return dict(self._levels)

    @property
    def aliases(self) -> Mapping[str, str]:

        return self._aliases

    @property
    def main(self) -> Optional[HandlerInfo]:

        if self._main is None:
            return None
        return HandlerInfo(
            hid=self._main.hid, level=self._main.level, fmt=self._main.fmt, path=None
        )

    @property
    def paths(self) -> AbstractSet[HandlerInfo]:

        if self._main is None:
            return set()
        return {HandlerInfo(hid=h.hid, level=h.level, fmt=h.fmt, path=h.sink) for h in self._paths}

    def config_levels(

        self,

        *,

        levels: Mapping[str, int] = _SENTINEL,

        colors: Mapping[str, str] = _SENTINEL,

        icons: Mapping[str, str] = _SENTINEL,

        aliases: Mapping[str, str] = _SENTINEL,

        add_methods: bool = True,

    ) -> __qualname__:

        levels = FancyLoguruDefaults.levels_extended if levels is _SENTINEL else levels
        colors = FancyLoguruDefaults.colors_extended if colors is _SENTINEL else colors
        icons = FancyLoguruDefaults.icons_extended if icons is _SENTINEL else icons
        aliases = FancyLoguruDefaults.aliases if aliases is _SENTINEL else aliases
        for k, v in levels.items():

            self.config_level(
                k,
                v,
                color=colors.get(k, _SENTINEL),
                icon=icons.get(k, _SENTINEL),
                add_methods=add_methods,
            )
        self._aliases = dict(aliases)
        return self

    def init(
        self,

        *,

        level: str = FancyLoguruDefaults.level,

        sink=sys.stderr,

        fmt: Formatter = FancyLoguruDefaults.fmt_built_in,

        intercept: bool = True,

    ) -> __qualname__:
        """
        Sets an initial configuration.
        """
        if intercept:
            # noinspection PyArgumentList
            logging.basicConfig(handlers=[InterceptHandler()], level=0, encoding="utf-8")
        logger.remove(None)  # get rid of the built-in handler
        self.config_main(level=level, sink=sink, fmt=fmt)
        return self

    def config_level(

        self,

        name: str,

        level: int,

        *,

        color: Union[None, str, _SENTINEL] = _SENTINEL,

        icon: Union[None, str, _SENTINEL] = _SENTINEL,

        add_methods: bool = True,

        replace: bool = True,

    ) -> __qualname__:
        try:
            data = logger.level(name)
        except ValueError:
            data = None
        if data is None:
            logger.level(
                name,
                no=level,
                color=None if color is _SENTINEL else color,
                icon=None if icon is _SENTINEL else icon,
            )
        elif replace:
            if level != data.no:  # loguru doesn't check whether they're eq; it just errors
                raise IllegalStateError(f"Cannot set level={level}!={data.no} for {name}")
            logger.level(
                name,
                color=data.color if color is _SENTINEL else color,
                icon=data.icon if icon is _SENTINEL else icon,
            )
        if add_methods:
            self._set_logger_levels([name])
        return self

    def _set_logger_levels(self, levels: Sequence[str]) -> __qualname__:
        for level in levels:

            def _x(__message: str, *args, **kwargs):
                logger.log(level, __message, *args, **kwargs)


            _x.__name__ = level.lower()
            if not hasattr(self._logger, level.lower()):
                setattr(self._logger, level.lower(), _x)
        return self

    def config_main(
        self,

        *,

        sink: TextIO = _SENTINEL,

        level: Optional[str] = _SENTINEL,

        fmt: Formatter = _SENTINEL,

    ) -> __qualname__:
        """
        Sets the logging level for the main handler (normally stderr).
        """
        if level is not None and level is not _SENTINEL:
            level = level.upper()
        if self._main is None:
            self._main = _HandlerInfo(
                hid=-1,
                sink=sys.stderr,
                level=self._levels[FancyLoguruDefaults.level],
                fmt=FancyLoguruDefaults.fmt_built_in,
            )
        else:
            try:
                logger.remove(self._main.hid)
            except ValueError:
                logger.error(f"Cannot remove handler {self._main.hid}")
        self._main.level = self._main.level if level is _SENTINEL else level
        self._main.sink = self._main.sink if sink is _SENTINEL else sink
        self._main.fmt = self._main.fmt if fmt is _SENTINEL else fmt
        self._main.hid = logger.add(self._main.sink, level=self._main.level, format=self._main.fmt)
        return self

    def remove_path(self, path: Path) -> __qualname__:

        for k, h in self._paths.items():

            if h.path.resolve() == path.resolve():
                h.level = None
                try:
                    logger.remove(k)
                except ValueError:
                    logger.error(f"Cannot remove handler {k} to {path}")

    def add_path(

        self,

        path: Path,

        level: str = _SENTINEL,

        *,

        fmt: str = _SENTINEL,

    ) -> __qualname__:
        if level is _SENTINEL and self._main is None:
            level = FancyLoguruDefaults.level
        elif level is _SENTINEL:
            level = self._main.level
        level = level.upper()
        if fmt is _SENTINEL and self._main is None:
            fmt = FancyLoguruDefaults.fmt_built_in
        elif fmt is _SENTINEL:
            fmt = self._main.fmt
        if isinstance(fmt, str):
            fmt = FormatFactory.with_extras(fmt=fmt)
        ell = self._levels[level]
        info = LogSinkInfo.guess(path)
        x = self._logger.add(

            str(info.base),
            format=fmt,
            level=level,
            compression=info.compression,
            serialize=info.serialize,
            backtrace=True,
            diagnose=True,
            enqueue=True,
            encoding="utf-8",
        )
        self._paths[x] = _HandlerInfo(hid=x, sink=info.base, level=ell, fmt=fmt)
        return self

    def from_cli(
        self,

        path: Union[None, str, Path] = None,

        main: Optional[str] = FancyLoguruDefaults.level,

    ) -> __qualname__:
        """
        This function controls logging set via command-line.

        Args:
            main: The level for stderr
            path: If set, the path to a file. Can be prefixed with ``:level:`` to set the level
                  (e.g. ``:INFO:mandos-run.log.gz``). Can serialize to JSON if .json is used
                  instead of .log or .txt.
        """
        if main is None:
            main = FancyLoguruDefaults.level
        main = self._aliases.get(main.upper(), main.upper())
        if main not in FancyLoguruDefaults.levels_extended:
            _permitted = ", ".join(
                [*FancyLoguruDefaults.levels_extended, *FancyLoguruDefaults.aliases.keys()]
            )
            raise XValueError(f"{main.lower()} not a permitted log level (allowed: {_permitted}")
        self.config_main(level=main)
        if path is not None or len(str(path)) == 0:
            match = _LOGGER_ARG_PATTERN.match(str(path))
            path_level = "DEBUG" if match.group(1) is None else match.group(1)
            path = Path(match.group(2))
            self.add_path(path, path_level)
            logger.info(f"Added logger to {path} at level {path_level}")
        logger.info(f"Set main log level to {main}")
        return self

    __call__ = from_cli


class LoggerWithNotice(Logger, metaclass=abc.ABCMeta):
    """
    A wrapper that has fake methods to trick static analysis.
    """

    def notice(self, __message: str, *args, **kwargs):

        raise NotImplementedError()  # not real


class LoggerWithCaution(Logger, metaclass=abc.ABCMeta):

    def caution(self, __message: str, *args, **kwargs):

        raise NotImplementedError()  # not real


class LoggerWithCautionAndNotice(LoggerWithNotice, LoggerWithCaution, metaclass=abc.ABCMeta):
    """
    A wrapper that has fake methods to trick static analysis.
    """


class FancyLoguruExtras:

    @classmethod
    def rewire_streams_to_utf8(cls) -> None:

        sys.stderr.reconfigure(encoding="utf-8")
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")

    @classmethod
    def built_in(

        cls,

        *,

        level: str = FancyLoguruDefaults.level,

        rewire: bool = False,

    ) -> FancyLoguru[Logger]:
        if rewire:
            cls.rewire_streams_to_utf8()
        return FancyLoguru.new(Logger).config_levels().init(level=level)

    @classmethod
    def extended(

        cls,

        *,

        level: str = FancyLoguruDefaults.level,

        simplify_fmt: bool = True,

        red_green_safe: bool = True,

        rewire: bool = False,

    ) -> FancyLoguru[LoggerWithCautionAndNotice]:
        if rewire:
            cls.rewire_streams_to_utf8()
        return (
            FancyLoguru.new(LoggerWithCautionAndNotice)
            .config_levels(
                levels=FancyLoguruDefaults.levels_extended,
                colors=(
                    FancyLoguruDefaults.colors_red_green_safe
                    if red_green_safe
                    else FancyLoguruDefaults.colors_extended
                ),
                icons=FancyLoguruDefaults.icons_extended,
            )
            .init(
                level=level,
                fmt=(
                    FancyLoguruDefaults.fmt_simplified
                    if simplify_fmt
                    else FancyLoguruDefaults.fmt_built_in
                ),
            )
        )


if __name__ == "__main__":
    _logger = FancyLoguruExtras.extended().logger
    _logger.caution("hello")


__all__ = [
    "FancyLoguruDefaults",
    "FancyLoguru",
    "FancyLoguruExtras",
    "HandlerInfo",
    "Logger",
    "LoggerWithCautionAndNotice",
]

1			"""
2			Loguru logging extension that, configurably:
3			- redirects built-in logging to your loguru logger
4			- remembers the handlers added
5			- auto-detects compression and serialization from filenames
6			- includes extras in non-serializing handlers
7			- has a few alternative (possibly better) choices for colors, icons, and levels
8			- will complain when you do really stupid things
9			- has a convenient notation for configuring from a CLI
10			(e.g. ``--stderr debug --log :INFO:run.log.gz``)
11			- mandes utf-8
12			"""
13			from __future__ import annotations
14
15			import abc
16			import logging
17			import os
18			import sys
19			from dataclasses import dataclass
20			from inspect import cleandoc
21			from pathlib import Path
22			from typing import (
23			AbstractSet,
24			Any,
25			Callable,
26			Generic,
27			Mapping,
28			Optional,
29			Sequence,
30			TextIO,
31			Type,
32			TypeVar,
33			Union,
34			)
35
36			import loguru._defaults as _defaults
			0 ignored issues – show introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report Unable to import 'loguru._defaults' Loading history...
37
38			# noinspection PyProtectedMember
39			import regex
dmyersturnbull / pocketutils

Push — main ( f8490e...e59e89 )

InterceptHandler.emit() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
