/haplog

Some utilities to enhance the experience of using the logging module

Primary LanguagePythonMIT LicenseMIT

haplog

⚠️ DEPRECATE: This project has been deprecated since loguru has most of the features that haplog originally intended to develop. Happy logging guys :)

Content

Introdoction

Provides enhancements for logging functionality, including:

  1. Color the texts of logging:

    haplog/haplog/logger_utils.py

    Lines 30 to 54 in 3146156

    # https://talyian.github.io/ansicolors/
    # black = "\x1b[30;20m"
    red = "\x1b[31;20m"
    bold_red = "\x1b[31;1m"
    green = "\x1b[32;20m"
    yellow = "\x1b[33;20m"
    # blue = "\x1b[34;20m"
    # magenta = "\x1b[35;20m"
    # cyan = "\x1b[36;20m"
    # white = "\x1b[37;20m"
    grey = "\x1b[38;20m"
    # \x1b[38;2;r;g;bm - foreground
    # \x1b[48;2;r;g;bm - background
    reset = "\x1b[0m"
    self.formats = {
    logging.DEBUG: grey + costom_format + reset,
    logging.INFO: green + costom_format + reset,
    logging.WARNING: yellow + costom_format + reset,
    logging.ERROR: red + costom_format + reset,
    logging.CRITICAL: bold_red + costom_format + reset,
    }
  2. Redirect the standard output of third-party modules to log records, which is usually used when developers are unwilling to spend time manually changing print() to the corresponding logging function.

    haplog/haplog/logger_utils.py

    Lines 62 to 68 in 3146156

    class OutputLogger:
    """
    serves as a pseudo file-like stream object that redirects written content to a logger instance.
    It overrides the `write` method to append the written messages to an internal buffer
    (`linebuf`). When a message ends with a newline character, it logs the buffered messages
    as a log record.
    """

    haplog/examples/demo_haplog.py

    Lines 30 to 34 in 3146156

    with redirect_stdout(
    OutputLogger(logger_name=LOGGER_NAME, logging_level=logging.DEBUG) # type: ignore
    ):
    print(MESSAGE + " by print()")
    third_party_function()
  3. Logging (of cource):

    haplog/haplog/logger_utils.py

    Lines 97 to 104 in 3146156

    class MultiProcessLogger:
    """
    Implements a custom logger designed for multi-process environments.
    It is based on the 'Logging Cookbook - Logging to a single file from multiple processes' example
    in the Python documentation. It utilizes a multiprocessing queue and a listener process to
    enable logging across multiple processes. The class provides functionality for logging records
    to a file and printing logs to the console.
    """
    1. single-process:

      haplog/examples/demo_haplog.py

      Lines 23 to 42 in 3146156

      def single_process():
      mpl = MultiProcessLogger(log_folder, level_console=logging.DEBUG)
      mpl.start()
      worker_configurer(mpl.queue) # type: ignore
      logger = logging.getLogger(LOGGER_NAME)
      with redirect_stdout(
      OutputLogger(logger_name=LOGGER_NAME, logging_level=logging.DEBUG) # type: ignore
      ):
      print(MESSAGE + " by print()")
      third_party_function()
      logger.debug(MESSAGE)
      logger.info(MESSAGE)
      logger.warning(MESSAGE)
      logger.error(MESSAGE)
      logger.critical(MESSAGE)
      mpl.join()
    2. multi-process:

      haplog/examples/demo_haplog.py

      Lines 58 to 66 in 3146156

      def multi_process():
      mpl = MultiProcessLogger(log_folder, level_console=logging.DEBUG)
      mpl.start()
      with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
      for _ in range(10):
      executor.submit(worker_process, mpl.queue, worker_configurer)
      mpl.join()

Installation

pip install "https://github.com/changchiyou/haplog/archive/main.zip"

Usage & Visual Effect

Download and execute /examples/demo_haplog.py:

haplog/examples/demo_haplog.py

Lines 1 to 71 in 3146156

import concurrent.futures
import logging
import logging.handlers
import multiprocessing
from contextlib import redirect_stdout
from pathlib import Path
from haplog import MultiProcessLogger, OutputLogger, worker_configurer
LOGGER_NAME = "test"
MESSAGE = "test"
log_folder = (Path(__file__).parent) / "logs"
if log_folder.exists() is False:
log_folder.mkdir()
def third_party_function():
print(MESSAGE + " by third_party_function()")
def single_process():
mpl = MultiProcessLogger(log_folder, level_console=logging.DEBUG)
mpl.start()
worker_configurer(mpl.queue) # type: ignore
logger = logging.getLogger(LOGGER_NAME)
with redirect_stdout(
OutputLogger(logger_name=LOGGER_NAME, logging_level=logging.DEBUG) # type: ignore
):
print(MESSAGE + " by print()")
third_party_function()
logger.debug(MESSAGE)
logger.info(MESSAGE)
logger.warning(MESSAGE)
logger.error(MESSAGE)
logger.critical(MESSAGE)
mpl.join()
def worker_process(queue, configurer):
import time
from random import random
configurer(queue)
name = multiprocessing.current_process().name
logger = logging.getLogger(name)
logger.info("Worker started: %s" % name)
time.sleep(random())
logger.info("Worker finished: %s" % name)
def multi_process():
mpl = MultiProcessLogger(log_folder, level_console=logging.DEBUG)
mpl.start()
with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
for _ in range(10):
executor.submit(worker_process, mpl.queue, worker_configurer)
mpl.join()
if __name__ == "__main__":
single_process()
multi_process()

image

Check ANSI Color Codes for more personalized color schemes.

Reference