feat: replace loguru with logging

shared/libretime_shared/cli.py

@@ -3,8 +3,6 @@ from typing import Any, Callable, Optional
 
 import click
 
-from .logging import INFO, LOG_LEVEL_MAP
-
 
 def cli_logging_options() -> Callable:
     def decorator(func: Callable) -> Callable:
@@ -18,8 +16,8 @@ def cli_logging_options() -> Callable:
         func = click.option(
             "--log-level",
             "log_level",
-            type=click.Choice(list(LOG_LEVEL_MAP.keys())),
-            default=INFO.name,
+            type=click.Choice(["error", "warning", "info", "debug"]),
+            default="info",
             help="Name of the logging level.",
         )(func)
 

shared/libretime_shared/config/_base.py

@@ -1,14 +1,16 @@
+import logging
 import sys
 from itertools import zip_longest
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
 
-from loguru import logger
 from pydantic import BaseModel, ValidationError
 from yaml import YAMLError, safe_load
 
 from ._env import EnvLoader
 
+logger = logging.getLogger(__name__)
+
 DEFAULT_ENV_PREFIX = "LIBRETIME"
 DEFAULT_CONFIG_FILEPATH = Path("/etc/libretime/config.yml")
 

shared/libretime_shared/logging.py

@@ -1,127 +1,38 @@
-import sys
-from copy import deepcopy
+import logging
+from logging.handlers import TimedRotatingFileHandler
 from pathlib import Path
-from typing import TYPE_CHECKING, NamedTuple, Optional, Tuple
+from typing import List, Optional, Tuple
 
-from loguru import logger
-
-if TYPE_CHECKING:
-    from loguru import Logger
-
-logger.remove()
-
-
-class LogLevel(NamedTuple):
-    name: str
-    no: int
-
-    def is_debug(self) -> bool:
-        return self.no <= 10
-
-
-# See https://loguru.readthedocs.io/en/stable/api/logger.html#levels
-ERROR = LogLevel(name="error", no=40)
-WARNING = LogLevel(name="warning", no=30)
-INFO = LogLevel(name="info", no=20)
-DEBUG = LogLevel(name="debug", no=10)
-TRACE = LogLevel(name="trace", no=5)
-
-LOG_LEVEL_MAP = {
-    ERROR.name: ERROR,
-    WARNING.name: WARNING,
-    INFO.name: INFO,
-    DEBUG.name: DEBUG,
-    TRACE.name: TRACE,
-}
-
-
-def level_from_name(name: str) -> LogLevel:
-    """
-    Find logging level, depending on the name provided.
-
-    :param name: name (one of "error", "warning", "info", "debug", "trace") of the log level
-    :returns: log level guessed from the name
-    :raises ValueError: on invalid level name
-    """
-    name = name.lower()
-    if name not in LOG_LEVEL_MAP:
-        raise ValueError(f"invalid level name '{name}'")
-    return LOG_LEVEL_MAP[name]
+logger = logging.getLogger(__name__)
 
 
 def setup_logger(
-    level: LogLevel,
+    level: str,
     filepath: Optional[Path] = None,
-    serialize: bool = False,
+    serialize: bool = False,  # pylint: disable=unused-argument
     rotate: bool = True,
-) -> Tuple[LogLevel, Optional[Path]]:
+) -> Tuple[str, Optional[Path]]:
     """
-    Configure the logger and return the computed log level.
-
-    See https://loguru.readthedocs.io/en/stable/overview.html
-
-    :param verbosity: verbosity (between -1 and 3) of the logger
-    :param filepath: write logs to filepath
-    :param serialize: generate JSON formatted log records
-    :param rotate: enable log rotation and retention
-    :returns: log level guessed from the verbosity
+    Configure the logger and return the log level and log filepath.
     """
-    handlers = [{"sink": sys.stderr, "level": level.no, "serialize": serialize}]
+    level = level.upper()
+
+    root = logging.getLogger()
+    root.setLevel(level)
+
+    formatter = logging.Formatter(
+        "%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)s - %(message)s"
+    )
+    handlers: List[logging.Handler] = [logging.StreamHandler()]
 
     if filepath is not None:
-        file_handler = {
-            "sink": filepath,
-            "enqueue": True,
-            "level": level.no,
-            "serialize": serialize,
-            "encoding": "utf-8",
-        }
         if rotate:
-            file_handler.update(
-                {
-                    "rotation": "12:00",
-                    "retention": "7 days",
-                    "compression": "gz",
-                }
-            )
+            handlers.append(TimedRotatingFileHandler(filepath, when="midnight"))
+        else:
+            handlers.append(logging.FileHandler(filepath))
 
-        handlers.append(file_handler)
-
-    logger.configure(handlers=handlers)
+    for handler in handlers:
+        handler.setFormatter(formatter)
+        root.addHandler(handler)
 
     return level, filepath
-
-
-_empty_logger = deepcopy(logger)
-
-
-def create_task_logger(
-    level: LogLevel,
-    filepath: Path,
-    serialize: bool = False,
-) -> "Logger":
-    """
-    Create and configure an independent logger for a task, return the new logger.
-
-    See #creating-independent-loggers-with-separate-set-of-handlers in
-    https://loguru.readthedocs.io/en/stable/resources/recipes.html
-
-    :returns: new logger
-    """
-    task_logger = deepcopy(_empty_logger)
-
-    task_logger.configure(
-        handlers=[
-            {
-                "sink": filepath,
-                "enqueue": True,
-                "level": level.no,
-                "serialize": serialize,
-                "rotation": "12:00",
-                "retention": "7 days",
-                "encoding": "utf-8",
-            }
-        ],
-    )
-
-    return task_logger