如何构建Python日志模块？实现单进程全局唯一Logger

我需要编写一个Python日志模块，要求Logger实例在每个进程中唯一且全局可用（无需作为参数在函数间传递），但在设计方案时遇到了瓶颈。

单进程初始实现 #

先从单场景入手，写出了单进程下的日志代码：

# process_1.py

from datetime import datetime
from datetime import timezone

import logging
import sys

logger_process_name = 'process_1'
logger_file_datetime = datetime.now(timezone.utc).date()

logger = logging.getLogger(__name__)

stdout_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

stdout_log_handler = logging.StreamHandler(stream=sys.stdout)
stdout_log_handler.setLevel(logging.INFO)
stdout_log_handler.setFormatter(stdout_log_formatter)

file_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

file_log_handler = logging.FileHandler(filename=f'{logger_process_name}_{logger_file_datetime}.log')
file_log_handler.setLevel(logging.DEBUG)
file_log_handler.setFormatter(file_log_formatter)

logger.setLevel(logging.DEBUG)
logger.addHandler(stdout_log_handler)
logger.addHandler(file_log_handler)

def main():
    logger.info('hello!')
    logger.info('goodbye!')

if __name__ == '__main__':
    main()

这段代码实现了同时输出日志到stdout和带UTC日期戳的文件，但存在大量初始化样板代码。

多进程下的重复代码问题 #

编写第二个进程process_2.py时，需要复制相同的日志初始化代码：

# process_2.py

from datetime import datetime
# ...
# ... 省略上述相同的日志初始化代码 ...
# ...
logger.addHandler(file_log_handler)

def main():
    logger.info('hello from process_2!')
    logger.info('goodbye from process_2!')

if __name__ == '__main__':
    main()

显然重复代码过多，需要将日志逻辑封装为独立模块。

初步封装的缺陷 #

我尝试编写了如下日志模块：

# lib_logger.py

from datetime import datetime
from datetime import timezone

import logging
import sys

logger_process_name = 'process_1' # <--- !!! 进程名错误 !!!
logger_file_datetime = datetime.now(timezone.utc).date()

logger = logging.getLogger(__name__)

stdout_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

stdout_log_handler = logging.StreamHandler(stream=sys.stdout)
stdout_log_handler.setLevel(logging.INFO)
stdout_log_handler.setFormatter(stdout_log_formatter)

file_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

file_log_handler = logging.FileHandler(filename=f'{logger_process_name}_{logger_file_datetime}.log')
file_log_handler.setLevel(logging.DEBUG)
file_log_handler.setFormatter(file_log_formatter)

logger.setLevel(logging.DEBUG)
logger.addHandler(stdout_log_handler)
logger.addHandler(file_log_handler)

但在process_2.py中使用该模块时，进程名和日志文件名都会出错：

# process_2.py

from lib_logger import logger

def main():
    logger.info('hello from process_2!') # <--- 此处打印错误的进程名
    logger.info('goodbye from process_2!') # <- 此处同样错误
    # 日志写入文件`process_1_2024-08-22.log`而非预期的`process_2_2024-08-22.log`

if __name__ == '__main__':
    main()

尝试的解决方案及问题 #

我尝试通过初始化函数和全局变量来解决问题：

# lib_logger.py

logger_process_name = None
logger = None

def set_logger_process_name(process_name: str) -> None:
    global logger_process_name
    logger_process_name = process_name

def get_logger() -> logging.Logger:
    global logger

    if logger is not None:
        return logger

    if logger_process_name is None:
        raise RuntimeError(f'process name not set, cannot create logger instance')

    logger = logging.getLogger(__name__)

    stdout_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

    stdout_log_handler = logging.StreamHandler(stream=sys.stdout)
    stdout_log_handler.setLevel(logging.INFO)
    stdout_log_handler.setFormatter(stdout_log_formatter)

    logger_file_datetime = datetime.now(timezone.utc).date()
    file_log_formatter = logging.Formatter('%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d | %(message)s')

    file_log_handler = logging.FileHandler(filename=f'{logger_process_name}_{logger_file_datetime}.log')
    file_log_handler.setLevel(logging.DEBUG)
    file_log_handler.setFormatter(file_log_formatter)

    logger.setLevel(logging.DEBUG)
    logger.addHandler(stdout_log_handler)
    logger.addHandler(file_log_handler)
    
    return logger

但这种方案存在导入顺序依赖的问题：

# process_1.py

from lib_logger import get_logger
from lib_logger import set_logger_process_name

# 进程名可能定义在其他模块
from lib_process_names import PROCESS_NAME_PROCESS_1

set_logger_process_name(PROCESS_NAME_PROCESS_1)
log = get_logger()

# 必须先初始化模块（至少调用set_logger_process_name）
# 再导入其他使用logger模块的模块（调用get_logger）
# 原因：调用get_logger前必须先调用set_logger_process_name
from example_module import example_function

比如example_module.py在导入时就会调用get_logger()，导致报错：

# example_module.py

from lib_logger import get_logger

log = get_logger()

def example_function():
    log.info('started work in example_function...')

虽然可以让所有函数在运行时才调用get_logger()，但这并非理想方案——Logger句柄应在模块内全局可达，不应让每个函数负责获取实例，这和传递Logger参数相比并无本质改进。

内容的提问来源于stack exchange，提问作者user2138149