缺漏待補：`pytest`單元測試

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
Installation
Usage & Visual Effect
Reference

Introdoction

Provides enhancements for logging functionality, including:

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, 

 }

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()

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. 

  """

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()

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()

Check ANSI Color Codes for more personalized color schemes.

	import logging
	from pathlib import Path

	from haplog import instantiate_logger

	LOGGER_NAME = 'test'
	MESSAGE = 'test'


	def test_console_info(capsys):
	instantiate_logger(LOGGER_NAME)
	logger = logging.getLogger(LOGGER_NAME)

	logger.info(MESSAGE)

	captured = capsys.readouterr()

	assert f'INFO {Path(__file__).name} - test_console_info() : {MESSAGE}' in captured.err


	def test_console_debug(capsys):
	instantiate_logger(LOGGER_NAME, level_console=logging.DEBUG)
	logger = logging.getLogger(LOGGER_NAME)

	logger.debug(MESSAGE)

	captured = capsys.readouterr()

	assert f'DEBUG {Path(__file__).name} - test_console_debug() : {MESSAGE}' in captured.err


	def test_console_debug_info_level(capsys):
	instantiate_logger(LOGGER_NAME)
	logger = logging.getLogger(LOGGER_NAME)

	logger.debug(MESSAGE)

	captured = capsys.readouterr()

	assert f'DEBUG {Path(__file__).name} - test_console_debug() : {MESSAGE}' not in captured.err


	def test_log(tmp_path, capsys):
	instantiate_logger(LOGGER_NAME, tmp_path)
	logger = logging.getLogger(LOGGER_NAME)

	logger.info(MESSAGE)
	logger.debug(MESSAGE)

	captured = capsys.readouterr()

	assert f'INFO {Path(__file__).name} - test_log() : {MESSAGE}' in captured.err
	assert f'DEBUG {Path(__file__).name} - test_log() : {MESSAGE}' not in captured.err

	File = open(tmp_path / 'record', mode='r')
	contents = File.read()
	File.close()

	assert f'INFO {Path(__file__).name} - test_log() : {MESSAGE}' in contents
	assert f'DEBUG {Path(__file__).name} - test_log() : {MESSAGE}' in contents

changchiyou / haplog Goto Github PK

haplog's Introduction

haplog

Content

Introdoction

Installation

Usage & Visual Effect

Reference

haplog's People

Contributors

Watchers

haplog's Issues

缺漏待補：`pytest`單元測試

新增 Single/Multi - Process 快捷使用

單元測試、文件更新

錯誤訊息洗頻

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent

	# 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,
	}

	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.
	"""

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

	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.
	"""

	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 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()

	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()