如何编写自定义Python日志处理器？

Question

如何编写一个自定义的控制台日志函数，使得在控制台窗口中输出的日志信息只在一行上显示（不追加），直到出现第一条正常的日志记录。

progress = ProgressConsoleHandler()
console  = logging.StreamHandler()  

logger = logging.getLogger('test')
logger.setLevel(logging.DEBUG) 
logger.addHandler(console)  
logger.addHandler(progress)

logger.info('test1')
for i in range(3):
    logger.progress('remaining %d seconds' % i)
    time.sleep(1)   
logger.info('test2')

这样控制台的输出就只有三行：

INFO: test1
remaining 0 seconds... 
INFO: test2

有没有什么好的建议，关于如何实现这个功能？

Answer 1

这是对以下内容的补充：自定义处理器，如果你决定使用json或yaml文件来存储你的日志配置（推荐这样做）。这样做的话，你可以在不需要创建子类的情况下自定义处理器，而且看起来会更清晰。

我在这里会添加一个我自己的例子，结合Python文档中的例子，应该能覆盖大部分情况：

假设你想给一个处理器设置一个自定义的日志级别，但标准库中的处理器并不支持这个功能。你可以使用一个简单的函数来自定义处理器的创建，比如：

import logging, logging.config

def your_handler_creator_function(stream=None): 
#The reason for this argument is that it's the one the __init__ method in the "logging.StreamHandler" class receives
    handler = logging.StreamHandler(stream) #The type of handler I want to use
    handler.setLevel(MY_CUSTOM_LEVEL_PREVIOUSLY_DEFINED)
    return handler

LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
    'default': {
        'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
    },
},
'handlers': {
    "name_of_your_handler": {
        "()": "path.to.your.your_handler_creator_function",
        # This value "stream" is the one that will be passed to the function
        "stream": "ext://sys.stdout",
        # The handler will get the values below automatically
        "formatter": "default",
        "filters": [
            "filter_name"
        ]
    },
},
'root': {
    'handlers': ['file'],
    'level': 'DEBUG',
},
}

logging.config.dictConfig(LOGGING)
logger = logging.getLogger('mylogger')
logger.debug('A debug message')

Answer 2

免责声明 1

我没有彻底测试这个解决方案，但它似乎可以满足基本的日志记录功能。而且，这种实现方式肯定不是最佳实践。因此，我不建议在生产环境中使用这个解决方案……至少在进一步测试之前不要使用。

免责声明 2

不推荐这种方法的部分原因是，你很容易在自己的包装类中不小心覆盖掉 logging.Logger 类的方法。如果发生这种情况，可能会导致奇怪且难以解释的错误。

如果使用这种方法，请务必检查你打算在自定义方法或属性中使用的任何名称，确保它们在 logging.Logger 类中不存在。

有关更多信息，请参考 logging 模块的文档。

不再啰嗦：答案来了

答案

我创建了一个自定义日志记录器的基础类，如下所示：

import logging

class CustomLoggerBC():
    def __init__(self, name:str):
         self._logger = logging.getLogger(name)
         newdict = {k: getattr(self._logger, k) for k in dir(self._logger) if k not in dir(self)}
         self.__dict__.update(newdict)

然后你可以写一个自己的自定义日志记录器，继承自基础类 CustomLoggerBC，如下所示：

class MyCustomLogger(CustomLoggerBC):
    def __init__(name:str, ...rest of your arguments here):
        super().__init__(name)
        ... rest of your custom constructor here

在你的类构造函数或类中的任何方法中，你可以对存储在 self._logger 中的 logging.Logger 实例进行任何想要的更改，或者通常会进行的更改，比如添加处理器或设置日志级别：

class MyCustomLogger(CustomLoggerBC):
    def __init__(name:str, ...rest of your arguments here):
        super().__init__(name)
        ... Define your handler here
        self._logger.addHandler(self._customHandler)
        self._logger.setLevel(logging.INFO)
        ... rest of your custom constructor here

示例

像这样的代码：

cl = MyCustomLogger("MyCustomLogger")
cl.info("This is a test")
cl.info("This is also a test")

很容易产生类似这样的输出：

[ INFO ] [05/19/2022 10:03:45] 这是一个测试

[ INFO ] [05/19/2022 10:03:45] 这也是一个测试

此外，你甚至可以覆盖 logging.Logger 类的方法（虽然我不推荐这样做）

class MyCustomLogger(CustomLoggerBC):
    def __init__(name:str, ...rest of your arguments here):
        super().__init__(name)
        ... rest of your custom constructor here
    def info(self, msg):
        print("My custom implementation of logging.Logger's info method!")

像这样的代码：

cl = MyCustomLogger("MyCustomLogger")
cl.info("This is a test")
cl.info("This is also a test")

现在会产生类似这样的输出：

我自定义的 logging.Logger 的 info 方法！

我自定义的 logging.Logger 的 info 方法！

为什么你想这样做？

你可以提前配置你的日志记录器。例如，如果你知道应用程序的 A 部分的日志记录器会有特定的格式、特定的处理器和特定的级别。在 A 部分做这样的事情会容易得多：

myPartALogger = PartACustomLogger()

而不是在 A 部分做所有日志记录器的初始化工作。

这也使得日志记录器更具可重用性。如果 A 部分和 B 部分需要不同的日志记录器，但配置相同（比如相同的级别和格式化器，但处理器不同），你可以为每个部分创建两个不同名称的 PartACustomLogger 实例，并为每个实例传递不同的处理器。

为什么它有效

本质上，你是在用 MyCustomLogger 包装 logging.Logger 类。基础类 CustomLoggerBC 更新了子类的实现实例字典，包含了 logging.Logger 类的所有方法和属性，这样实现就基本上可以作为 logging.Logger 对象来使用。

对你的自定义日志记录器类 MyCustomLogger 的任何属性或方法请求，看起来都是转发给 logging.Logger 实例，由 logging.Logger 实例来处理，但实际上是你的类在处理这些请求。这使得看起来像是在子类化 logging.Logger 类，但实际上并不是。

Answer 3

import logging
class ProgressConsoleHandler(logging.StreamHandler):
    """
    A handler class which allows the cursor to stay on
    one line for selected messages
    """
    on_same_line = False
    def emit(self, record):
        try:
            msg = self.format(record)
            stream = self.stream
            same_line = hasattr(record, 'same_line')
            if self.on_same_line and not same_line:
                stream.write(self.terminator)
            stream.write(msg)
            if same_line:
                stream.write('... ')
                self.on_same_line = True
            else:
                stream.write(self.terminator)
                self.on_same_line = False
            self.flush()
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self.handleError(record)

if __name__ == '__main__':
    import time
    progress = ProgressConsoleHandler()
    console  = logging.StreamHandler()  

    logger = logging.getLogger('test')
    logger.setLevel(logging.DEBUG) 
    logger.addHandler(progress)

    logger.info('test1')
    for i in range(3):
        logger.info('remaining %d seconds', i, extra={'same_line':True})
        time.sleep(1)   
    logger.info('test2')

注意，这里只注册了一个处理器，并且使用了extra这个关键字参数来告诉处理器它应该保持在一行上。emit()方法里面还有更多的逻辑，用来处理哪些消息应该保持在一行，哪些消息需要单独占一行。