如何实现仅需初始化一次的CustomLogger,类同原生logging?
解决CustomLogger全局复用问题
问题现状
当前实现的CustomLogger需要在每个文件中重复实例化(传入tracker ID等参数)才能使用,无法像原生logging一样在一个入口文件初始化后,其他文件直接调用。
解决方案思路
利用Python模块的单例特性,将Logger实例封装为模块级别的全局对象,提供初始化函数和获取函数,确保整个程序中只有一个Logger实例,且只需在入口文件初始化一次。
修改后的代码实现
1. 重构CustomLogger.py
调整为全局实例模式,增加初始化和获取方法:
import json import logging import os class StreamLogger(logging.Logger): def __init__(self, name, level=logging.DEBUG): super().__init__(name, level) streamHandler = logging.StreamHandler() formatter = logging.Formatter('[%(asctime)s] [%(funcName)s] %(message)s') streamHandler.setFormatter(formatter) self.addHandler(streamHandler) # 全局Logger实例,初始为None _global_logger = None def setup(name, k_dict, level=logging.DEBUG): """初始化全局Logger,仅第一次调用有效""" global _global_logger if _global_logger is None: class Logger(logging.Logger): def __init__(self, name, k_dict, level=logging.DEBUG): super().__init__(name, level) self.k_dict = k_dict.copy() # 复制字典避免外部修改影响日志 self.json_logging = 'TRANSFORM_TO_JSON' in os.environ self.stream_logger = StreamLogger(name, level) def _transform_to_json(self, msg): msg_dict = self.k_dict.copy() msg_dict['message'] = msg return json.dumps(msg_dict) def info(self, msg, *args, **kwargs): if self.json_logging: self.stream_logger.info(self._transform_to_json(msg), *args, **kwargs) else: self.stream_logger.info(msg, *args, **kwargs) _global_logger = Logger(name, k_dict, level) return _global_logger def get_logger(): """获取已初始化的全局Logger""" if _global_logger is None: raise RuntimeError("Logger未初始化,请先调用setup()方法") return _global_logger
2. 修改入口文件test-executor.py
仅在入口处调用setup初始化Logger:
import CustomLogger from CustomExecutor import iterative_executor def main(): # 初始化全局Logger logging = CustomLogger.setup('sample-run', {'cnt': 10}) logging.info("Starting run") iterative_executor(1) logging.info("Done with execution") if __name__ == '__main__': main()
3. 修改CustomExecutor.py
直接获取全局Logger实例,无需重复初始化:
from CustomLogger import get_logger # 获取已初始化的全局Logger logging = get_logger() def iterative_executor(iter_cnt=2): for i in range(iter_cnt): logging.info(f"Current iteration cnt {i}")
关键优化点
- 模块级变量
_global_logger确保全局只有一个Logger实例,避免重复初始化 setup函数仅在第一次调用时创建实例,后续调用直接返回已有实例- 对
k_dict进行复制,防止外部字典修改影响日志内容 - 修复了原代码中
json_transform函数名与调用不匹配的问题
内容的提问来源于stack exchange,提问作者madmatrix
相关产品推荐
相关产品推荐

