禁用主日志时为多个子日志文件设置 Python 日志记录

Question

我有一些相关但独立的 Python 脚本，它们都使用两个使用日志记录的内部模块。

第一个脚本使用根记录器工作正常，并从两个模块捕获日志语句。然而，对于第二个脚本，我想要一个主日志，但是当它遍历服务器列表时，将日志发送到每台机器的日志文件，同时暂停记录到主日志文件和控制台。我现在有一个 hacky 解决方案，我将在下面展示。

import logging

DEFAULT_LOG_FORMAT = "%(asctime)s [%(levelname)s]: %(message)s"
DEFAULT_LOG_LEVEL = logging.INFO

def get_log_file_handler(filename, level=None, log_format=None):
  file_handler = logging.FileHandler(filename=filename, encoding="utf-8", mode="w")
  file_handler.setLevel(level or DEFAULT_LOG_LEVEL)
  file_handler.setFormatter(logging.Formatter(log_format or DEFAULT_LOG_FORMAT))

  return file_handler

def process(server):
  server_file_handler = get_log_file_handler("%s.log" % server.name)
  root_logger = logging.getLogger()

  # This works, but is hacky
  main_handlers = list(root_logger.handlers) # copy list of root log handlers
  root_logger.handlers = [] # empty the list on the root logger

  root_logger.addHandler(server_file_handler)

  try:
    # do some stuff with the server
    logging.info("This should show up only in the server-specific log file.")
  finally:
    root_logger.removeHandler(server_file_handler)

    # Add handlers back in
    for handler in main_handlers:
      root_logger.addHandler(handler)

def main():
  logging.basicConfig(level=DEFAULT_LOG_LEVEL)

  logging.getLogger().addHandler(get_log_file_handler("main.log"))

  servers = [] # retrieved from another function, just here for iteration

  logging.info("This should show up in the console and main.log.")

  for server in servers:
    process(server)

  logging.info("This should show up in the console and main.log again.")


if __name__ == "__main__":
  main()

我正在寻找一种不那么棘手的方法来做到这一点。我意识到仅仅调用 logging.info() 和类似的是一个问题，并且已经更改了两个模块中的代码以使用:

logger = logging.getLogger("moduleA")

和

logger = logging.getLogger("moduleB")

因此主脚本，无论是 scriptA.py 还是 scriptB.py，都使用根记录器，将从这两个模块获取事件，并将其传播并记录到 main.log。我尝试过的其他一些解决方案是在所有现有处理程序上使用过滤器，这些处理程序会忽略“moduleA”和“moduleB”中的所有内容。

我的下一个想法是为各个服务器创建一个新的命名记录器，将 server_file_handler 作为它们的唯一处理程序，并将其添加为两个模块记录器的处理程序，并在流程结束时删除这些处理程序().然后我可以将根记录器的级别设置为警告，这样来自两个模块的所有 INFO/DEBUG 语句将只转到特定于服务器的记录器。

我不能完全使用分层记录器命名，除非它以某种方式支持通配符，因为我最终会得到:

logging.getLogger("org.company") # main logger for script
logging.getLogger("org.company.serverA") 
logging.getLogger("org.company.serverB")
logging.getLogger("org.company.moduleA")
logging.getLogger("org.company.moduleB")

来自两个模块的日志只会传播到主记录器，而不是两个服务器日志。

这基本上是一个他们期望一棵树，我需要一个图的问题。以前有人做过这样的事情吗？最符合 Pythonic 的方法是什么？

Answer 1

这是一个有趣的问题。我的第一直觉是使用 logger.getChild 但默认实现不会执行您想要的操作。假设您可以动态地将处理程序添加到单个记录器，它仍然不会执行您想要的操作，因为您必须向主文件处理程序和服务器处理程序添加过滤器以过滤不应进入服务器的消息日志和反之亦然。

也就是说，好消息是，为每个 child 创建处理程序的自定义记录器实际上非常简单，可以通过修改 getChild 的简单子(monad)类来完成，仅此而已。

下面的大变化只是 HandlerPerChildLogger，Logger 与普通 Logger 的不同之处在于它需要两个参数，而不仅仅是一个 name 参数。

import logging

DEFAULT_LOG_FORMAT = "%(asctime)s [%(levelname)s]: %(message)s"
DEFAULT_LOG_LEVEL = logging.INFO

class HandlerPerChildLogger(logging.Logger):
    selector = "server"

    def __init__(self, name, handler_factory, level=logging.NOTSET):
        super(HandlerPerChildLogger, self).__init__(name, level=level)
        self.handler_factory = handler_factory

    def getChild(self, suffix):
        logger = super(HandlerPerChildLogger, self).getChild(suffix)
        if not logger.handlers:
            logger.addHandler(self.handler_factory(logger.name))
            logger.setLevel(DEFAULT_LOG_LEVEL)
        return logger

def file_handler_factory(name):
    handler = logging.FileHandler(filename="{}.log".format(name), encoding="utf-8", mode="a")
    formatter = logging.Formatter(DEFAULT_LOG_FORMAT)
    handler.setFormatter(formatter)
    return handler

logger = HandlerPerChildLogger("my.company", file_handler_factory)
logger.setLevel(DEFAULT_LOG_LEVEL)
ch = logging.StreamHandler()
fh = logging.FileHandler(filename="my.company.log", encoding="utf-8", mode="a")
ch.setLevel(DEFAULT_LOG_LEVEL)
fh.setLevel(DEFAULT_LOG_LEVEL)
formatter = logging.Formatter(DEFAULT_LOG_FORMAT)
ch.setFormatter(formatter)
fh.setFormatter(formatter)
logger.addHandler(ch)
logger.addHandler(fh)

def process(server):
    server_logger = logger.getChild(server)
    server_logger.info("This should show up only in the server-specific log file for %s", server)
    server_logger.info("another log message for %s", server)

def main():
    # servers list retrieved from another function, just here for iteration
    servers = ["server1", "server2", "server3"]

    logger.info("This should show up in the console and main.log.")

    for server in servers:
        process(server)

    logger.info("This should show up in the console and main.log again.")

if __name__ == "__main__":
    main()