mandos.model.utils.fancy_logger

Complexity

Total Complexity

55

Size/Duplication

Total Lines	396
Duplicated Lines	0 %

Importance

Changes

0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	LogSinkInfo.guess()	0	17	4
B	FancyLoguru.config_levels()	0	16	6
A	FormatFactory.plain()	0	6	1
A	FancyLoguru.__init__()	0	6	1
A	FancyLoguru.main()	0	6	2
A	FancyLoguru.paths()	0	5	2
A	InterceptHandler.emit()	0	12	3
A	FancyLoguru.remove_path()	0	8	4
A	LoguruUtils.force_streams_to_utf8()	0	6	1
B	FancyLoguru.config_main()	0	26	8
A	FancyLoguru.logger()	0	3	1
C	FancyLoguru.config_level()	0	29	9
A	FancyLoguru.aliases()	0	3	1
A	FancyLoguru.init()	0	16	2
B	FancyLoguru.__call__()	0	29	6
A	FancyLoguru.levels()	0	3	1
A	FancyLoguru.add_path()	0	19	1
A	FormatFactory.with_extras()	0	15	2

from __future__ import annotations

Missing module docstring

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

# noinspection PyProtectedMember
import loguru._defaults as _defaults

Unable to import 'loguru._defaults'

import regex

Unable to import 'regex'

from loguru import logger

Unable to import 'loguru'

# noinspection PyProtectedMember
from loguru._logger import Logger

Unable to import 'loguru._logger'

from pocketutils.core.exceptions import IllegalStateError, XValueError

Unable to import 'pocketutils.core.exceptions'

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:

Missing class docstring

    @classmethod
    def with_extras(

Missing function or method docstring

        cls,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        fmt: str = _FMT,

Wrong hanging indentation before block (add 4 spaces).

        sep: str = "; ",

Wrong hanging indentation before block (add 4 spaces).

        eq_sign: str = " ",

Wrong hanging indentation before block (add 4 spaces).

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

Function name "FMT" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            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]:

Missing function or method docstring

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

The argument record seems to be unused.

Function name "FMT" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            return fmt.replace("[EXTRA]", "")

        return FMT


class _SENTINEL:
    pass


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

Class name "T" doesn't conform to PascalCase naming style ('[^\\W\\da-z][^\\W_]+$' pattern)

_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 Defaults:

Missing class docstring

    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 = dict(
        TRACE="<dim>",
        DEBUG="<dim>",
        INFO="<bold>",
        CAUTION="<yellow>",
        SUCCESS="<blue>",
        WARNING="<yellow>",
        NOTICE="<blue>",
        ERROR="<red>",
        CRITICAL="<red>",
    )
    icons = dict(
        TRACE=_defaults.LOGURU_TRACE_ICON,
        DEBUG=_defaults.LOGURU_DEBUG_ICON,
        INFO=_defaults.LOGURU_INFO_ICON,
        CAUTION="⚐",  # or ☡
        SUCCESS=_defaults.LOGURU_SUCCESS_ICON,
        WARNING=_defaults.LOGURU_WARNING_ICON,
        NOTICE="★",
        ERROR=_defaults.LOGURU_ERROR_ICON,
        CRITICAL=_defaults.LOGURU_CRITICAL_ICON,
    )

    level: str = "INFO"

    fmt = FormatFactory.with_extras()

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


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

Missing class docstring

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)

Missing class docstring

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

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

Missing function or method docstring

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

Variable name "c" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            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]):

Missing class docstring

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

    @property
    def logger(self) -> T:

Missing function or method docstring

        return self._logger

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

Missing function or method docstring

        return dict(self._levels)

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

Missing function or method docstring

        return self._aliases

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

Missing function or method docstring

        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]:

Missing function or method docstring

        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(

Missing function or method docstring

        self,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        levels: Mapping[str, int] = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

        colors: Mapping[str, str] = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

        icons: Mapping[str, str] = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

        aliases: Mapping[str, str] = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

    ) -> __qualname__:

Undefined variable '__qualname__'

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

Variable name "v" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

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

    def init(
        self,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        level: str = Defaults.level,

Wrong hanging indentation before block (add 4 spaces).

        sink=sys.stderr,

Wrong hanging indentation before block (add 4 spaces).

        fmt: Formatter = Defaults.fmt,

Wrong hanging indentation before block (add 4 spaces).

        intercept: bool = True,

Wrong hanging indentation before block (add 4 spaces).

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

    def config_level(

Missing function or method docstring

        self,

Wrong hanging indentation before block (add 4 spaces).

        name: str,

Wrong hanging indentation before block (add 4 spaces).

        level: int,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

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

Wrong hanging indentation before block (add 4 spaces).

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

Wrong hanging indentation before block (add 4 spaces).

        replace: bool = True,

Wrong hanging indentation before block (add 4 spaces).

    ) -> __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,
            )
        return self

    def config_main(
        self,

Wrong hanging indentation before block (add 4 spaces).

        *,

Wrong hanging indentation before block (add 4 spaces).

        sink: TextIO = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

        level: Optional[str] = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

        fmt: Formatter = _SENTINEL,

Wrong hanging indentation before block (add 4 spaces).

    ) -> __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[Defaults.level], fmt=Defaults.fmt
            )
        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__:

Missing function or method docstring

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

Variable name "h" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            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(

Missing function or method docstring

        self, path: Path, level: str = Defaults.level, *, fmt: str = Defaults.fmt

Wrong hanging indentation before block (add 4 spaces).

    ) -> __qualname__:
        level = level.upper()
        ell = self._levels[level]
        info = LogSinkInfo.guess(path)
        x = logger.add(

Variable name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            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 __call__(
        self,

Wrong hanging indentation before block (add 4 spaces).

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

Wrong hanging indentation before block (add 4 spaces).

        main: Optional[str] = Defaults.level,

Wrong hanging indentation before block (add 4 spaces).

    ) -> __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 = Defaults.level
        main = self._aliases.get(main.upper(), main.upper())
        if main not in Defaults.levels_extended:
            _permitted = ", ".join([*Defaults.levels_extended, *Defaults.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)
            self.logger.info(f"Added logger to {path} at level {path_level}")
        self.logger.info(f"Set main log level to {main}")
        return self


class LoguruUtils:

Missing class docstring

    @classmethod
    def force_streams_to_utf8(cls) -> None:

Missing function or method docstring

        # we warn the user about this in the docs!
        sys.stderr.reconfigure(encoding="utf-8")
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")


__all__ = ["Defaults", "FancyLoguru", "LoguruUtils", "HandlerInfo"]