Commit c49ee675 authored by Bishnoi, Bhaskar's avatar Bishnoi, Bhaskar
Browse files

creates different logger (rich, loguru, python logging

parent 2829e190
Loading
Loading
Loading
Loading
+2 −2
Original line number Diff line number Diff line
version: '3.7'
version: "3.7"

services:
  package:
+169 −20
Original line number Diff line number Diff line
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import sys
"""Provide a standard logger for all use cases."""
import logging
# standard python imports
import os

from rich.logging import RichHandler
from rich.traceback import install


def create_logger(file_out=None, mode=None, encoding=None):
def create_handler(file_out, mode, encoding, another_handler=None):
    """
    create_logger:
        Creates a logger for all uses
    create_handler:
        Creates a logging handler to format text written by Python logging module.
    
    :param file_out: Path to file to output logs to, defaults to None
    :type file_out: str, optional
@@ -20,27 +18,57 @@ def create_logger(file_out=None, mode=None, encoding=None):
    :type mode: str, optional
    :param encoding: Encoding to open the file with, defaults to None
    :type encoding: str, optional
    :return: The logger that was created
    :rtype: Logger
    :param another_handler: Handler from another logging module, defaults to None
    :type another_handler: handler, optional
    :return: A Python loggging handler
    :rtype: handler
    """
    # Remove any handler's that may have been set in the logging root
    for handler in logging.root.handlers[:]:
        handler.close()
        logging.root.removeHandler(handler)

    install()
    log_level = os.environ.get("LOGLEVEL", "INFO").upper()
    rich_handler = RichHandler(rich_tracebacks=True, markup=True)
    if file_out is not None:
        file_handler = logging.FileHandler(file_out, mode, encoding)
        file_handler_fmt = logging.Formatter(
            "[%(asctime)s]" + "%(levelname)8s - " + " - %(message)s"
        )
        file_handler.setFormatter(file_handler_fmt)
        handlers = [rich_handler, file_handler]
        handlers = [file_handler]
        if another_handler: handlers = [another_handler, file_handler]
    else:
        file_handler = None
        handlers = [rich_handler]
        if another_handler: 
            handlers = [another_handler] 
        else: handlers = [] 
    
    return file_handler, handlers


def create_rich_logger(file_out=None, 
                       mode=None, 
                       encoding=None):
    """
    create_rich_logger:
        Creates a Rich logger for all uses
    
    :param file_out: Path to file to output logs to, defaults to None
    :type file_out: str, optional
    :param mode: Mode to open the file with, defaults to None
    :type mode: str, optional
    :param encoding: Encoding to open the file with, defaults to None
    :type encoding: str, optional
    :return: A Rich logger instance
    :rtype: Logger
    """

    try:
        from rich.logging import RichHandler
        from rich.console import Console
        from rich.traceback import install
    except ImportError:
        raise ImportError("Failed to load rich logger")

    install()
    log_level = os.environ.get("LOGLEVEL", "INFO").upper()
    rich_handler = RichHandler(rich_tracebacks=True, markup=True, console=Console(stderr=True))
    file_handler, handlers = create_handler(file_out, mode, encoding, rich_handler)
    logging.basicConfig(
        level=log_level,
        format="%(message)s",
@@ -53,3 +81,124 @@ def create_logger(file_out=None, mode=None, encoding=None):
        file_handler.close()
    rich_handler.close()
    return logging.getLogger("rich")


def create_loguru_logger(file_out=None,
                        mode=None, 
                        encoding=None):
    """
    create_loguru_logger:
        Creates a Rich logger for all uses
    
    :param file_out: Path to file to output logs to, defaults to None
    :type file_out: str, optional
    :param mode: Mode to open the file with, defaults to None
    :type mode: str, optional
    :param encoding: Encoding to open the file with, defaults to None
    :type encoding: str, optional
    :return: A Loguru logger instance
    :rtype: Logger
    """

    try:
        from loguru import logger as loguru_logger
    except:
        print("Failed to load loguru logger")
        return

    loguru_logger.remove()
    log_level = os.environ.get("LOGLEVEL", "INFO").upper()
    if file_out is not None:
        loguru_logger.add(
            file_out,
            mode=mode,
            encoding=encoding,
            level=log_level,
            format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}",
            backtrace=True,
            diagnose=True
        )
    else:
        loguru_logger.add(
            sys.stderr,
            level=log_level,
            format="[{time:YYYY/MM/DD HH:mm:ss}] - {level} - {message}",
            backtrace=True,
            diagnose=True
        )
    return loguru_logger


def create_python_logger(file_out=None,
                        mode=None, 
                        encoding=None):
    """
    create_python_logger:
        Creates a Python logger for all uses
    
    :param file_out: Path to file to output logs to, defaults to None
    :type file_out: str, optional
    :param mode: Mode to open the file with, defaults to None
    :type mode: str, optional
    :param encoding: Encoding to open the file with, defaults to None
    :type encoding: str, optional
    :return: Python logger instance
    :rtype: Logger
    """

    log_level = os.environ.get("LOGLEVEL", "INFO").upper()
    file_handler, handlers = create_handler(file_out, mode, encoding)
    if file_handler:
        logging.basicConfig(
            level=log_level,
            format="%(message)s",
            datefmt="[%Y/%m/%d %H:%M;%S]",
            handlers=handlers,
        )
    else:
        logging.basicConfig(
            level=log_level,
            format="%(asctime)s" + "%(levelname)8s - " + " - %(message)s",
            datefmt="[%Y/%m/%d %H:%M;%S]"
        )
    # if there is a file_handler set, close it before leaving :)
    # This prevents leaving open files
    if file_handler is not None:
        file_handler.close()
    return logging.getLogger()


def create_logger(file_out=None, 
                  mode=None, 
                  encoding=None,
                  logger_type="rich"):
    """
    create_logger:
        Creates a logger for all uses
    
    :param file_out: Path to file to output logs to, defaults to None
    :type file_out: str, optional
    :param mode: Mode to open the file with, defaults to None
    :type mode: str, optional
    :param encoding: Encoding to open the file with, defaults to None
    :type encoding: str, optional
    :param logger_type: Logger to use for logging purpose, defaults to rich (rich (default), loguru, logging)
    :type logger_type: str, optional
    :return: The logger that was created
    :rtype: Logger
    """
    # Remove any handler's that may have been set in the logging root
    for handler in logging.root.handlers[:]:
        handler.close()
        logging.root.removeHandler(handler)

    logger_type = os.getenv("LOGGER_TYPE", None) or logger_type
    mode = mode or "a"
    encoding = encoding or "utf-8"

    if logger_type == "loguru":
        return create_loguru_logger(file_out, mode, encoding)
    elif logger_type == "logging":
        return create_python_logger(file_out, mode, encoding)
    
    return create_rich_logger(file_out, mode, encoding)
+2 −1
Original line number Diff line number Diff line
@@ -30,7 +30,8 @@ classifiers = [
    "Topic :: Software Development :: Build Tools"
]
dependencies = [
    "rich==13.7.1"
    "rich==13.7.1",
    "loguru==0.7.2"
]

[project.optional-dependencies]
+2 −1
Original line number Diff line number Diff line
@@ -4,3 +4,4 @@ psycopg2-binary==2.9.9
pymssql==2.2.11
SQLAlchemy~=2.0.23
influxdb==5.3.1
loguru==0.7.2
 No newline at end of file

test/test_logz.py

0 → 100644
+79 −0
Original line number Diff line number Diff line
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import unittest
import os
import io
import sys
from common.logz import create_logger
from common.file_operations import path_exists, read_file, delete_file

class TestLogger(unittest.TestCase):
    
    def setUp(self):
        self.stream_io = io.StringIO()
        self.system_stdr = sys.stderr
        sys.stderr = self.stream_io
        
    def tearDown(self):
        sys.stderr = self.system_stdr
        # remove log files
        log_files = ["test_rich.log", "test_loguru.log", "test_python_logging.log"]
        for file in log_files:
            if (path_exists(file)):
                delete_file(file)

    def test_rich_logger(self):
        log_file = "test_rich.log"
        logger = create_logger(file_out=log_file)
        custom_rich_text = "Testing rich logger 1"
        logger.info(custom_rich_text)

        stderr_output = self.stream_io.getvalue()
        self.assertIn(custom_rich_text, stderr_output)


        file_contents = read_file(log_file)
        if file_contents == None:
            return f"An error occured with rich file {log_file}"
        self.assertIn(custom_rich_text, file_contents)

    def test_loguru(self):
        log_file = "test_loguru.log"
        logger = create_logger(logger_type="logger")
        custom_loguru_text = "Testing Loguru logger 2"
        logger.info(custom_loguru_text)

        stderr_output = self.stream_io.getvalue()
        self.assertIn(custom_loguru_text, stderr_output)
        logger = None

        logger = create_logger(file_out=log_file, logger_type="logger")
        custom_loguru_text = "Testing Loguru logger 2.1"
        logger.info(custom_loguru_text)
        file_contents = read_file(log_file)
        if file_contents == None:
            return f"An error occured with loguru file {log_file}"
        self.assertIn(custom_loguru_text, file_contents)

    def test_python_logging(self):
        log_file = "test_python_logging.log"
        logger = create_logger(logger_type="logging")
        custom_python_logging_text = "Testing Python Logging logger 3"
        logger.info(custom_python_logging_text)

        stderr_output = self.stream_io.getvalue()
        self.assertIn(custom_python_logging_text, stderr_output)
        logger = None


        logger = create_logger(file_out=log_file, logger_type="logging")
        custom_python_logging_text = "Testing Loguru logger 3.1"
        logger.info(custom_python_logging_text)
        file_contents = read_file(log_file)
        if file_contents == None:
            return f"An error occured with Python logging file {log_file}"
        self.assertIn(custom_python_logging_text, file_contents)


if __name__ == "__main__":
    unittest.main(verbosity=2)