Passed
Push — main ( 83a9fb...fa90c4 )
by Douglas
03:43
created

MandosLogging.configure_level()   C

Complexity

Conditions 9

Size

Total Lines 29
Code Lines 26

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 9
eloc 26
nop 7
dl 0
loc 29
rs 6.6666
c 0
b 0
f 0
1
from __future__ import annotations
0 ignored issues
show
introduced by
Missing module docstring
Loading history...
2
import logging
3
import sys
4
from dataclasses import dataclass
5
from inspect import cleandoc
6
from pathlib import Path
7
from typing import Mapping, Optional, Union
8
9
# noinspection PyProtectedMember
10
import loguru._defaults as _defaults
0 ignored issues
show
introduced by
Unable to import 'loguru._defaults'
Loading history...
11
import regex
0 ignored issues
show
introduced by
Unable to import 'regex'
Loading history...
12
from loguru import logger
0 ignored issues
show
introduced by
Unable to import 'loguru'
Loading history...
13
14
# noinspection PyProtectedMember
15
from loguru._logger import Logger
0 ignored issues
show
introduced by
Unable to import 'loguru._logger'
Loading history...
16
from pocketutils.core.exceptions import IllegalStateError, XValueError
0 ignored issues
show
introduced by
Unable to import 'pocketutils.core.exceptions'
Loading history...
17
18
_LOGGER_ARG_PATTERN = regex.compile(r"(?:([a-zA-Z]+):)?(.*)", flags=regex.V1)
19
_DEFAULT_LEVEL: str = "INFO"
20
FMT = cleandoc(
21
    r"""
22
    <bold>{time:YYYY-MM-DD HH:mm:ss.SSS}</bold> |
23
    <level>{level: <8}</level> |
24
    <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>
25
    """
26
).replace("\n", " ")
27
28
29
class _SENTINEL:
30
    pass
31
32
33
log_compressions = {
34
    ".xz",
35
    ".lzma",
36
    ".gz",
37
    ".zip",
38
    ".bz2",
39
    ".tar",
40
    ".tar.gz",
41
    ".tar.bz2",
42
    ".tar.xz",
43
}
44
valid_log_suffixes = {
45
    *{f".log{c}" for c in log_compressions},
46
    *{f".txt{c}" for c in log_compressions},
47
    *{f".json{c}" for c in log_compressions},
48
}
49
# the levels for caution and notice are DEFINED here
50
# trace and success must match loguru's
51
# and the rest must match logging's
52
# note that most of these alternate between informative and problematic
53
# i.e. info (ok), caution (bad), success (ok), warning (bad), notice (ok), error (bad)
54
LEVELS = dict(
55
    TRACE=_defaults.LOGURU_TRACE_NO,
56
    DEBUG=_defaults.LOGURU_DEBUG_NO,
57
    INFO=_defaults.LOGURU_INFO_NO,
58
    CAUTION=23,
59
    SUCCESS=25,
60
    WARNING=_defaults.LOGURU_WARNING_NO,
61
    NOTICE=35,
62
    ERROR=_defaults.LOGURU_ERROR_NO,
63
    CRITICAL=_defaults.LOGURU_CRITICAL_NO,
64
)
65
LEVEL_COLORS = dict(
66
    TRACE="",
67
    DEBUG="",
68
    INFO="<bold>",
69
    CAUTION="<yellow>",
70
    SUCCESS="<blue>",
71
    WARNING="<yellow>",
72
    NOTICE="<blue>",
73
    ERROR="<red>",
74
    CRITICAL="<RED>",
75
)
76
LEVEL_ICONS = {}
77
_ALIASES = dict(NONE="OFF", NO="OFF", VERBOSE="INFO", QUIET="ERROR")
78
79
80
@dataclass(frozen=True, repr=True, order=True)
0 ignored issues
show
introduced by
Missing class docstring
Loading history...
81
class LogSinkInfo:
82
    path: Path
83
    base: Path
84
    suffix: str
85
    serialize: bool
86
    compression: Optional[str]
87
88
    @classmethod
89
    def guess(cls, path: Union[str, Path]) -> LogSinkInfo:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
90
        path = Path(path)
91
        base, compression = path.name, None
92
        for c in log_compressions:
0 ignored issues
show
Coding Style Naming introduced by
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)

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
93
            if path.name.endswith(c):
94
                base, compression = path.name[: -len(c)], c
95
        if not [base.endswith(s) for s in [".json", ".log", ".txt"]]:
96
            raise XValueError(
97
                f"Log filename {path.name} is not .json, .log, .txt, or a compressed variant"
98
            )
99
        return LogSinkInfo(
100
            path=path,
101
            base=path.parent / base,
102
            suffix=compression,
103
            serialize=base.endswith(".json"),
104
            compression=compression,
105
        )
106
107
108
class InterceptHandler(logging.Handler):
109
    """
110
    Redirects standard logging to loguru.
111
    """
112
113
    def emit(self, record):
114
        # Get corresponding Loguru level if it exists
115
        try:
116
            level = logger.level(record.levelname).name
117
        except ValueError:
118
            level = record.levelno
119
        # Find caller from where originated the logged message
120
        frame, depth = logging.currentframe(), 2
121
        while frame.f_code.co_filename == logging.__file__:
122
            frame = frame.f_back
123
            depth += 1
124
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
125
126
127
def _notice(__message: str, *args, **kwargs):
128
    return logger.log("NOTICE", __message, *args, **kwargs)
129
130
131
def _caution(__message: str, *args, **kwargs):
132
    return logger.log("CAUTION", __message, *args, **kwargs)
133
134
135
class MyLogger(Logger):
136
    """
137
    A wrapper that has a fake notice() method to trick static analysis.
138
    """
139
140
    def __init__(self, *args, **kwargs):
141
        super().__init__(*args, **kwargs)
142
        self._current_stderr_log_level: Optional[str] = _DEFAULT_LEVEL
143
144
    @property
145
    def current_stderr_log_level(self) -> Optional[str]:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
146
        return self._current_stderr_log_level
147
148
    def notice(self, __message: str, *args, **kwargs):
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
149
        raise NotImplementedError()  # not real
150
151
    def caution(self, __message: str, *args, **kwargs):
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
152
        raise NotImplementedError()  # not real
153
154
155
class MandosLogging:
0 ignored issues
show
introduced by
Missing class docstring
Loading history...
156
    # this is required for mandos to run
157
158
    DEFAULT_LEVEL: str = _DEFAULT_LEVEL
159
    _main_handler_id: int = None
160
161
    @classmethod
162
    def inverse_levels(cls) -> Mapping[int, str]:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
163
        return {v: k for k, v in cls.levels()}
164
165
    @classmethod
166
    def levels(cls) -> Mapping[str, int]:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
167
        return dict(LEVELS)
168
169
    @classmethod
170
    def init(
171
        cls,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
172
        *,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
173
        level: str = _DEFAULT_LEVEL,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
174
        sink=sys.stderr,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
175
        force_utf8: bool = True,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
176
        intercept: bool = True,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
177
        force: bool = False,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
178
    ) -> None:
179
        """
180
        Sets an initial configuration.
181
        """
182
        if cls._main_handler_id is not None and not force:
183
            return
184
        if force_utf8:
185
            # we warn the user about this in the docs!
186
            sys.stderr.reconfigure(encoding="utf-8")
187
            sys.stdout.reconfigure(encoding="utf-8")
188
            sys.stdin.reconfigure(encoding="utf-8")
189
        if intercept:
190
            logging.basicConfig(handlers=[InterceptHandler()], level=0, encoding="utf-8")
191
        for k, v in LEVELS.items():
0 ignored issues
show
Coding Style Naming introduced by
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)

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
192
            cls.configure_level(
193
                k, v, color=LEVEL_COLORS.get(k, _SENTINEL), icon=LEVEL_ICONS.get(k, _SENTINEL)
194
            )
195
        logger.notice = _notice
196
        logger.caution = _caution
197
        logger.remove(None)  # get rid of the built-in handler
198
        cls.set_main_level(level, sink=sink)
199
        logger.disable("chembl_webresource_client")
200
        logger.notice("Started.")
201
202
    @classmethod
203
    def configure_level(
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
204
        cls,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
205
        name: str,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
206
        level: int,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
207
        *,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
208
        color: Union[None, str, _SENTINEL] = _SENTINEL,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
209
        icon: Union[None, str, _SENTINEL] = _SENTINEL,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
210
        replace: bool = True,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
211
    ):
212
        try:
213
            data = logger.level(name)
214
        except ValueError:
215
            data = None
216
        if data:
217
            if replace:
218
                if level != data.no:  # loguru doesn't check whether they're eq; it just errors
219
                    raise IllegalStateError(f"Cannot set level={level}!={data.no} for {name}")
220
                logger.level(
221
                    name,
222
                    color=data.color if color is _SENTINEL else color,
223
                    icon=data.icon if icon is _SENTINEL else icon,
224
                )
225
        else:
226
            logger.level(
227
                name,
228
                no=level,
229
                color=None if color is _SENTINEL else color,
230
                icon=None if icon is _SENTINEL else icon,
231
            )
232
233
    @classmethod
234
    def set_main_level(cls, level: str, sink=sys.stderr) -> None:
235
        """
236
        Sets the logging level for the main handler (normally stderr).
237
        """
238
        level = level.upper()
239
        if cls._main_handler_id is not None:
240
            try:
241
                logger.remove(cls._main_handler_id)
242
            except ValueError:
243
                logger.error(f"Cannot remove handler {cls._main_handler_id}")
244
        if level != "OFF":
245
            cls._main_handler_id = logger.add(sink, level=level, format=FMT)
246
247
    @classmethod
248
    def add_path(cls, path: Path, level: str) -> int:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
249
        level = level.upper()
250
        info = LogSinkInfo.guess(path)
251
        return logger.add(
252
            str(info.base),
253
            level=level,
254
            compression=info.compression,
255
            serialize=info.serialize,
256
            backtrace=True,
257
            diagnose=True,
258
            enqueue=True,
259
            encoding="utf-8",
260
        )
261
262
263
class MandosSetup:
0 ignored issues
show
introduced by
Missing class docstring
Loading history...
264
    @classmethod
265
    def aliases(cls) -> Mapping[str, str]:
0 ignored issues
show
introduced by
Missing function or method docstring
Loading history...
266
        return _ALIASES
267
268
    def __call__(
269
        self,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
270
        log: Union[None, str, Path] = None,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
271
        level: Optional[str] = MandosLogging.DEFAULT_LEVEL,
0 ignored issues
show
Coding Style introduced by
Wrong hanging indentation before block (add 4 spaces).
Loading history...
272
    ) -> None:
273
        """
274
        This function controls all aspects of the logging as set via command-line.
275
276
        Args:
277
            level: The level for stderr
278
            log: If set, the path to a file. Can be prefixed with ``:level:`` to set the level
279
                  (e.g. ``:INFO:mandos-run.log.gz``). Can serialize to JSON if .json is used
280
                  instead of .log or .txt.
281
        """
282
        if level is None:
283
            level = MandosLogging.DEFAULT_LEVEL
284
        level = level.upper()
285
        level = _ALIASES.get(level, level)
286
        if level not in LEVELS:
287
            _permitted = ", ".join([*LEVELS, *_ALIASES.keys()])
288
            raise XValueError(f"{level.lower()} not a permitted log level (allowed: {_permitted}")
289
        MandosLogging.set_main_level(level)
290
        if log is not None or len(str(log)) == 0:
291
            match = _LOGGER_ARG_PATTERN.match(str(log))
292
            path_level = "DEBUG" if match.group(1) is None else match.group(1)
293
            path = Path(match.group(2))
294
            MandosLogging.add_path(path, path_level)
295
            logger.info(f"Added logger to {path} at level {path_level}")
296
        logger.info(f"Set log level to {level}")
297
298
299
# weird as hell, but it works
300
# noinspection PyTypeChecker
301
logger: MyLogger = logger
302
MandosLogging.init()
303
304
305
MANDOS_SETUP = MandosSetup()
306
307
__all__ = ["logger", "MANDOS_SETUP", "MandosLogging", "LogSinkInfo"]
308