FastAPI中Python logging模块日志重复问题求助

问题根源 #

每次请求/start接口时，都会实例化LogClass，而get_logger方法每次都会给**根日志器（logging.getLogger()）**添加新的日志处理器（Handler）。日志器的处理器是累积的：

• 第一次请求：添加1个StreamHandler，日志打印1次
• 第二次请求：再添加1个StreamHandler，总共2个，日志打印2次
• 第三次请求：再添加1个，总共3个，日志打印3次

这就是日志重复的核心原因。另外直接使用根日志器也容易和FastAPI/uvicorn自带的日志系统产生冲突。

修复方案 #

方案1：应用启动时初始化日志（推荐） #

将日志初始化逻辑放在应用启动阶段，避免每次请求重复创建日志实例。

修改后的完整代码：

import uvicorn
from fastapi import FastAPI

import logging
from logging.handlers import RotatingFileHandler, SysLogHandler
from logging import StreamHandler


# 全局初始化日志，仅执行一次
def setup_logging(
    log_format: str = "%(asctime)s - [%(levelname)s] - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s",
    log_name: str = None,
    maxFileSize: int = 1000000,
    backupCount: int = 3,
    syslog_host: str = None,
    syslog_port: int = 514,
    stream: bool = True
) -> logging.Logger:
    # 使用命名日志器，避免与根日志器冲突
    logger = logging.getLogger("fastapi_app")
    # 若已有处理器，直接返回，避免重复添加
    if logger.handlers:
        return logger
    
    logger.setLevel(logging.INFO)
    formatter = logging.Formatter(log_format)

    if log_name is not None:
        file_handler = RotatingFileHandler(log_name, maxBytes=maxFileSize, backupCount=backupCount)
        file_handler.setLevel(logging.INFO)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
    
    if syslog_host is not None and syslog_port is not None:
        syslog_handler = SysLogHandler(address=(syslog_host, syslog_port))
        syslog_handler.setLevel(logging.INFO)
        syslog_handler.setFormatter(formatter)
        logger.addHandler(syslog_handler)
    
    if stream:
        stream_handler = StreamHandler()
        stream_handler.setLevel(logging.INFO)
        stream_handler.setFormatter(formatter)
        logger.addHandler(stream_handler)
    
    return logger


# 启动时完成日志初始化
logger = setup_logging()

app = FastAPI()


@app.get("/start")
async def start_func():
    logger.info("Message to log")
    return {"result": "OK"}


if __name__ == "__main__":
    uvicorn.run("test2_app:app", host="127.0.0.1", port=8000, reload=True)

方案2：将LogClass改为单例模式 #

如果需要保留类的形式，通过单例模式限制LogClass仅实例化一次，确保处理器只添加一次：

import uvicorn
from fastapi import FastAPI

import logging
from logging.handlers import RotatingFileHandler, SysLogHandler
from logging import StreamHandler


class LogClass:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self,
                 log_format: str = "%(asctime)s - [%(levelname)s] - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s",
                 log_name: str = None,
                 maxFileSize: int = 1000000,
                 backupCount: int = 3,
                 syslog_host: str = None,
                 syslog_port: int = 514):
        if self._initialized:
            return
        
        self.log_backupCount = backupCount
        self.log_maxFileSize = maxFileSize
        self.log_format = log_format
        self.syslog_host = syslog_host
        self.syslog_port = syslog_port
        self.log_name = log_name

        self.logger = self.get_logger()
        self._initialized = True

    def get_file_handler(self):
        file_handler = RotatingFileHandler(self.log_name, maxBytes=self.log_maxFileSize,
                                           backupCount=self.log_backupCount)
        file_handler.setLevel(logging.INFO)
        file_handler.setFormatter(logging.Formatter(self.log_format))
        return file_handler

    def get_stream_handler(self):
        stream_handler = StreamHandler()
        stream_handler.setLevel(logging.INFO)
        stream_handler.setFormatter(logging.Formatter(self.log_format))
        return stream_handler

    def get_syslog_handler(self):
        syslog_handler = SysLogHandler(address=(self.syslog_host, self.syslog_port))
        syslog_handler.setLevel(logging.INFO)
        syslog_handler.setFormatter(logging.Formatter(self.log_format))
        return syslog_handler

    def get_logger(self, stream: bool = True) -> logging.Logger:
        # 使用命名日志器，避免与根日志器冲突
        logger = logging.getLogger("fastapi_app")
        logger.setLevel(logging.INFO)

        # 仅当无处理器时添加，避免重复
        if not logger.handlers:
            if self.log_name is not None:
                logger.addHandler(self.get_file_handler())
            if self.syslog_host is not None and self.syslog_port is not None:
                logger.addHandler(self.get_syslog_handler())
            if stream:
                logger.addHandler(self.get_stream_handler())

        return logger


app = FastAPI()


@app.get("/start")
async def start_func():
    log = LogClass()
    logger = log.logger
    logger.info("Message to log")

    return {"result": "OK"}


if __name__ == "__main__":
    uvicorn.run("test2_app:app", host="127.0.0.1", port=8000, reload=True)

关键改进点 #

• 使用命名日志器替代根日志器，避免与FastAPI/uvicorn的日志系统冲突
• 确保日志处理器仅添加一次：要么通过判断logger.handlers是否为空，要么通过单例模式限制实例化次数
• 将日志初始化逻辑移到应用启动阶段，而非每次请求执行

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