feat: configurator of loggers

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")

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)

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