From 25d196f56d26e88230426f4bfc94b231b5e173d4 Mon Sep 17 00:00:00 2001 From: dd84ai Date: Sun, 28 Jan 2024 02:28:08 +0100 Subject: [PATCH] feat: configurator of loggers --- .vscode/settings.json | 7 --- examples/test_examples.py | 28 +++++++++-- logus/logcore.py | 30 +++++++---- logus/utils.py | 101 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 146 insertions(+), 20 deletions(-) delete mode 100644 .vscode/settings.json create mode 100644 logus/utils.py diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index 597079a..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,7 +0,0 @@ -{ - "editor.formatOnSave": true, - "terminal.integrated.env.linux": { - "LOGUS_LOG_LEVEL": "DEBUG", - "LOGUS_LOG_JSON": "true" - } -} \ No newline at end of file diff --git a/examples/test_examples.py b/examples/test_examples.py index ae49f7b..6e0a3c5 100644 --- a/examples/test_examples.py +++ b/examples/test_examples.py @@ -1,15 +1,37 @@ import unittest from logus import get_logger +import logus from . import app_logus from . import app_types +from logus.utils import Loggers, LibName, LogLevel +import logging logger = get_logger(__name__) class TestExamples(unittest.TestCase): - def test_basic(self): - logger.warn("Writing something", app_logus.TaskID(123)) - def test_another_one(self): + def setUp(self) -> None: + Loggers( + root_log_level=logging.DEBUG, + log_levels={ + LibName("examples"): LogLevel(logging.DEBUG), + }, + add_time=True, + ).configure() + + def test_basic(self) -> None: + logger.warn("Writing something", app_logus.TaskID(app_types.TaskID(123))) + + def test_another_one(self) -> None: task = app_types.Task(smth="abc", b=4) logger.warn("Writing something", app_logus.Task(task)) + + def test_with_fields(self) -> None: + + logger2 = logger.with_fields(app_logus.Task(app_types.Task(smth="aaa", b=1))) + logger3 = logger.with_fields(logus.String("smth", "asd"), logus.Int("number", 2)) + + logger.info("logger printed") + logger2.info("logger2 printed") + logger3.info("logger3 printed") diff --git a/logus/logcore.py b/logus/logcore.py index bf49237..35c8825 100644 --- a/logus/logcore.py +++ b/logus/logcore.py @@ -1,6 +1,7 @@ import json import logging -from typing import Dict +from typing import Dict, List, Optional +from copy import deepcopy from . import settings from .logtypes import LogOption, LogParams @@ -15,15 +16,17 @@ class StructuredMessage: def __init__( self, message: str, + turn_json: Optional[bool], *args: LogOption, ) -> None: self._message = message + self._turn_json = turn_json self._kwargs: LogParams = LogParams({}) for add_option in args: add_option(self._kwargs) def __str__(self) -> str: - if settings.LOG_JSON: + if settings.LOG_JSON or self._turn_json: return json.dumps(self.to_dict()) if len(self._kwargs.keys()) == 0: @@ -42,28 +45,35 @@ def to_dict(self) -> Dict: class Logus: - def __init__(self, name: str): + def __init__(self, name: str, turn_json: Optional[bool]): """ pass __file__ into file """ self.logger = logging.getLogger(name) self.logger.setLevel(default_log_level) + self._with_fields: List[LogOption] = [] + self._turn_json = turn_json def debug(self, message: str, *args: LogOption) -> None: - self.logger.debug(StructuredMessage(message, *args)) + self.logger.debug(StructuredMessage(message,self._turn_json, *args)) def info(self, message: str, *args: LogOption) -> None: - self.logger.info(StructuredMessage(message, *args)) + self.logger.info(StructuredMessage(message, self._turn_json, *args)) def warn(self, message: str, *args: LogOption) -> None: - self.logger.warning(StructuredMessage(message, *args)) + self.logger.warning(StructuredMessage(message,self._turn_json, *args)) def error(self, message: str, *args: LogOption) -> None: - self.logger.error(StructuredMessage(message, *args)) + self.logger.error(StructuredMessage(message,self._turn_json, *args)) def fatal(self, message: str, *args: LogOption) -> None: - self.logger.fatal(StructuredMessage(message, *args)) + self.logger.fatal(StructuredMessage(message,self._turn_json, *args)) + def with_fields(self, *args: LogOption) -> "Logus": + logger = deepcopy(self) + logger._with_fields.extend(args) + return logger -def get_logger(file: str) -> Logus: - return Logus(file) + +def get_logger(file: str, turn_json: Optional[bool] = None) -> Logus: + return Logus(file, turn_json) diff --git a/logus/utils.py b/logus/utils.py new file mode 100644 index 0000000..d3a060a --- /dev/null +++ b/logus/utils.py @@ -0,0 +1,101 @@ +import logging +from . import settings +from typing import NewType, Dict, Optional, List +from dataclasses import dataclass, field +import json +import os + +LibName = NewType("LibName", str) +LogLevel = NewType("LogLevel", int) + +is_configured: bool = False + +@dataclass +class Loggers: + root_log_level: int + log_levels: Dict[LibName, LogLevel] = field(default_factory=dict) + turn_json: bool = False + + add_thread: bool = os.environ.get("LOGUS_ADD_THREAD")== "true" + add_process: bool = os.environ.get("LOGUS_ADD_PROCESS")== "true" + add_level: bool = os.environ.get("LOGUS_ADD_LEVEL")== "true" + add_filepath: bool = os.environ.get("LOGUS_ADD_FILEPATH")== "true" + add_time: bool = os.environ.get("LOGUS_ADD_TIME")== "true" + + @property + def _is_turn_json(self) -> bool: + return settings.LOG_JSON or self.turn_json + + @property + def _format_json(self) -> str: + message_format = "%(message)s" + format = {"content": "TARGET_REPLACE"} + if self.add_process: + format["process"] ="%(process)d" + if self.add_thread: + format["thread"] ="%(thread)d" + if self.add_time: + format["time"] ="%(asctime)s" + if self.add_filepath: + format["filepath"] ="%(name)s" + if self.add_level: + format["level"] ="%(levelname)s" + + + return json.dumps(format).replace('"TARGET_REPLACE\"',message_format) + + @property + def _format_text(self) -> str: + formats: List[str] = [] + if self.add_process: + formats.append("%(process)d") + if self.add_thread: + formats.append("%(thread)d") + if self.add_time: + formats.append("%(asctime)s") + if self.add_filepath: + formats.append("%(name)s") + if self.add_level: + formats.append("%(levelname)s") + + formats.append(" %(message)s") + return ":".join(formats) + + def configure(self) -> None: + """ + * third party libs are noisy and having bad default log levels + * for better logging purposes we should disable all other loggining to Warning level + * and turn on our app logging Debug level. + * it helps to be better aware about warnings and critical errors across libraries + * And at the same having very comfortable development environment which + makes very easy to investigate throughly our app debugging log records + and to fix from third party libs warnings only + """ + print("Configured debugging logging") + + global is_configured + + if is_configured: + return + + for lib_name, log_level in self.log_levels.items(): + loggers = [ + logging.getLogger(name) + for name in logging.root.manager.loggerDict + if name.startswith(lib_name) + ] + for logger in loggers: + logger.setLevel(log_level) + + root_logger = logging.getLogger("") + root_logger.setLevel(self.root_log_level) + ch = logging.StreamHandler() + ch.setLevel(self.root_log_level) + if self._is_turn_json: + formatter = logging.Formatter(self._format_json) + else: + formatter = logging.Formatter(self._format_text) + ch.setFormatter(formatter) + root_logger.addHandler(ch) + + is_configured = True