feat: logging

logger/__init__.py

@@ -0,0 +1 @@
+from logger.builder import logger_builder as logger_builder

logger/builder.py

@@ -0,0 +1,60 @@
+import logging
+import logging.config
+
+from logger.constants import (
+    LEVEL_NAME,
+    BACKUP_COUNT,
+    LOGS_FOLDER,
+    MAX_LOG_FILE_SIZE_BYTES,
+)
+from logger.formatter import JsonFormatter
+from logger.gunzip_rotating_file_handler import GunZipRotatingFileHandler
+from logger.filters import LevelFilter, RequestIdFilter
+from core.singleton import Singleton
+
+
+class LoggerBuilder(metaclass=Singleton):
+    def get_logger(self) -> logging.Logger:
+        logger = logging.getLogger("crm")
+        logger.setLevel(logging.DEBUG)
+        logger.handlers.clear()
+        self.set_handlers(logger)
+        return logger
+
+    def set_handlers(self, logger: logging.Logger):
+        LOGGER_LEVEL_STEP = 10
+
+        for level in range(logging.DEBUG, logging.CRITICAL + 1, LOGGER_LEVEL_STEP):
+            logger.addHandler(self.create_rotating_file_handler(level))
+
+        logger.addHandler(self.create_console_handler())
+
+    @classmethod
+    def create_rotating_file_handler(cls, level: int) -> GunZipRotatingFileHandler:
+        folder = LOGS_FOLDER / LEVEL_NAME[level]
+        folder.mkdir(parents=True, exist_ok=True)
+        filename = LEVEL_NAME[level] + ".log"
+
+        file_handler = GunZipRotatingFileHandler(
+            folder / filename,
+            maxBytes=MAX_LOG_FILE_SIZE_BYTES,
+            encoding="UTF-8",
+            backupCount=BACKUP_COUNT[level],
+        )
+
+        file_handler.addFilter(LevelFilter(level))
+        file_handler.addFilter(RequestIdFilter())
+        file_handler.setFormatter(JsonFormatter())
+
+        return file_handler
+
+    @classmethod
+    def create_console_handler(cls) -> logging.StreamHandler:
+        console_handler = logging.StreamHandler()
+        console_handler.setLevel(logging.INFO)
+
+        console_handler.setFormatter(JsonFormatter())
+        return console_handler
+
+
+logger_builder = LoggerBuilder()

logger/constants.py

@@ -0,0 +1,25 @@
+import logging
+from pathlib import Path
+
+from constants import APP_PATH
+
+LEVEL_NAME = {
+    logging.FATAL: "fatal",
+    logging.CRITICAL: "critical",
+    logging.ERROR: "error",
+    logging.WARNING: "warning",
+    logging.INFO: "info",
+    logging.DEBUG: "debug",
+}
+
+BACKUP_COUNT = {
+    logging.FATAL: 5,
+    logging.CRITICAL: 5,
+    logging.ERROR: 4,
+    logging.WARNING: 3,
+    logging.INFO: 2,
+    logging.DEBUG: 1,
+}
+
+MAX_LOG_FILE_SIZE_BYTES = 30 * 1024 * 1024  # 30 Mb
+LOGS_FOLDER = Path(APP_PATH) / Path("logs")

logger/filters.py

@@ -0,0 +1,17 @@
+import logging
+import uuid
+
+
+class LevelFilter(logging.Filter):
+    def __init__(self, level):
+        super().__init__()
+        self.level = level
+
+    def filter(self, record):
+        return record.levelno == self.level
+
+
+class RequestIdFilter(logging.Filter):
+    def filter(self, record):
+        record.request_id = str(uuid.uuid4())
+        return True

logger/formatter.py

@@ -0,0 +1,20 @@
+import json
+import logging
+from datetime import datetime, UTC
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record: any):
+        log_record = {
+            "timestamp": datetime.now(UTC).isoformat(),
+            "level": record.levelname,
+            "module": record.module,
+            "line": record.lineno,
+            "message": record.getMessage(),
+            "request_id": record.request_id,
+        }
+
+        if record.exc_info:
+            log_record["exception"] = self.formatException(record.exc_info)
+
+        return json.dumps(log_record)

logger/gunzip_rotating_file_handler.py

@@ -0,0 +1,35 @@
+import gzip
+from logging.handlers import RotatingFileHandler
+import shutil
+import os
+
+
+class GunZipRotatingFileHandler(RotatingFileHandler):
+    def doRollover(self):
+        if self.stream:
+            self.stream.close()
+            self.stream = None
+
+        if self.backupCount > 0:
+            # Rotate existing backup files
+            for i in range(self.backupCount - 1, 0, -1):
+                sfn = self.rotation_filename("%s.%d.gz" % (self.baseFilename, i))
+                dfn = self.rotation_filename("%s.%d.gz" % (self.baseFilename, i + 1))
+                if os.path.exists(sfn):
+                    if os.path.exists(dfn):
+                        os.remove(dfn)
+                    os.rename(sfn, dfn)
+
+            # Compress current log file to .1.gz
+            dfn = self.rotation_filename(self.baseFilename + ".1.gz")
+            if os.path.exists(dfn):
+                os.remove(dfn)
+
+            if os.path.exists(self.baseFilename):
+                with open(self.baseFilename, "rb") as f_in:
+                    with gzip.open(dfn, "wb") as f_out:
+                        shutil.copyfileobj(f_in, f_out)
+                os.remove(self.baseFilename)
+
+        if not self.delay:
+            self.stream = self._open()