mypydantic.helpers.logger

  1# pylint: disable = super-with-arguments,broad-exception-caught
  2import decimal
  3import json
  4import logging
  5import os
  6from datetime import datetime
  7from typing import Optional, Union
  8
  9import coloredlogs
 10
 11
 12class JsonHelper(json.JSONEncoder):
 13    def default(self, o):
 14        if isinstance(o, decimal.Decimal):
 15            return int(o)
 16        if isinstance(o, set):
 17            return list(o)
 18        if isinstance(o, bytes):
 19            return str(o)
 20        if isinstance(o, datetime):
 21            return str(o)
 22        return super(JsonHelper, self).default(o)
 23
 24
 25class CustomLogger:
 26    log_level = os.getenv("LOG_LEVEL", "INFO")
 27    local = False
 28    propogate = False
 29
 30    def log(self):
 31        logger = logging.getLogger(__name__)
 32        logger.handlers = []
 33        if self.log_level == "ERROR":
 34            coloredlogs.DEFAULT_FIELD_STYLES = {
 35                "asctime": {"bold": True, "color": "black"},
 36                "levelname": {"bold": True, "color": "red"},
 37            }
 38        elif self.log_level == "WARNING":
 39            coloredlogs.DEFAULT_FIELD_STYLES = {
 40                "asctime": {"bold": True, "color": "black"},
 41                "levelname": {"bold": True, "color": "yellow"},
 42            }
 43        elif self.log_level == "DEBUG":
 44            coloredlogs.DEFAULT_FIELD_STYLES = {
 45                "asctime": {"bold": True, "color": "black"},
 46                "levelname": {"bold": True, "color": "blue"},
 47            }
 48        else:
 49            coloredlogs.DEFAULT_FIELD_STYLES = {
 50                "asctime": {"bold": True, "color": "black"},
 51                "levelname": {"bold": True, "color": "blue"},
 52            }
 53        coloredlogs.DEFAULT_LOG_FORMAT = "%(levelname)s: %(asctime)s | %(message)s"
 54
 55        coloredlogs.install(level=self.log_level, logger=logger)
 56        logger.propagate = self.propogate
 57        return logger
 58
 59    def get_message(self, message, details: Optional[Union[list, dict]] = None):
 60        try:
 61            if isinstance(message, (list, dict)):
 62                if self.local:
 63                    message = json.dumps(message, indent=4, cls=JsonHelper)
 64                else:
 65                    message = json.dumps(message, cls=JsonHelper)
 66            if details:
 67                if message[-1] != ":":
 68                    message = f"{message}:"
 69                if isinstance(details, (list, dict)):
 70                    if self.local:
 71                        details = json.dumps(details, indent=4, cls=JsonHelper)
 72                    else:
 73                        details = json.dumps(details, cls=JsonHelper)
 74                message = f"{message}\n{details}"
 75            return message
 76        except Exception:
 77            return message
 78
 79    def info(self, message, details: Optional[Union[list, dict]] = None):
 80        message = self.get_message(message, details)
 81        self.log().info("%s", message)
 82
 83    def debug(self, message, details: Optional[Union[list, dict]] = None):
 84        message = self.get_message(message, details)
 85        self.log().debug("%s", message)
 86
 87    def warning(self, message, details: Optional[Union[list, dict]] = None):
 88        message = self.get_message(message, details)
 89        self.log().warning("%s", message)
 90
 91    def error(self, message, details: Optional[Union[list, dict]] = None):
 92        message = self.get_message(message, details)
 93        self.log().error("%s", message)
 94
 95    def output(self, values: dict):
 96        wrkdir = os.getcwd()
 97        file_path = "output_logs.json"
 98        with open(file_path, "w", encoding="utf-8") as out_file:
 99            out_file.write(json.dumps(values, indent=4, cls=JsonHelper))
100        self.log().info("Logs output to: %s/%s", wrkdir, file_path)