使用Gunicorn搭配自定义getMessage()日志时出现日志错误

我在Dash应用中自定义了LogRecord类来处理日志中的session_id变量，代码如下：

class LogRecord(logging.LogRecord):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.session_id = None

    def getMessage(self):
        msg = str(self.msg)
        if self.args:
            msg = msg % self.args
            self.session_id = self.args[0]
        return msg


logging.setLogRecordFactory(LogRecord)

直接用Dash启动应用时日志功能正常，但使用Gunicorn启动时，Gunicorn复用了自定义LogRecord却因自身日志参数格式不同触发错误，错误栈如下：

File ".../lib/python3.9/logging/__init__.py", line 1083, in emit
    msg = self.format(record)
  File ".../lib/python3.9/logging/__init__.py", line 927, in format
    return fmt.format(record)
  File ".../lib/python3.9/logging/__init__.py", line 663, in format
    record.message = record.getMessage()
  File ".../LipViz/log/classes_log.py", line 17, in getMessage
    self.session_id = self.args[0]
  File ".../lib/python3.9/site-packages/gunicorn/glogging.py", line 108, in __getitem__
    if k.startswith("{"):
AttributeError: 'int' object has no attribute 'startswith'

问题原因 #

Gunicorn自身日志使用了自定义参数数组，与应用日志的参数格式不兼容，复用全局替换的LogRecord后，自定义getMessage()逻辑处理Gunicorn日志时触发类型错误。

解决方案 #

核心是区分日志来源，让Gunicorn日志使用原生getMessage()，应用日志保留自定义逻辑，以下是两种可行实现：

方式一：基于日志名称前缀做分支处理

Gunicorn的日志器名称默认以gunicorn.开头，在自定义getMessage()中判断日志名称，针对性执行逻辑：

import logging

class LogRecord(logging.LogRecord):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.session_id = None

    def getMessage(self):
        # Gunicorn日志直接调用原生实现
        if self.name.startswith("gunicorn."):
            return super().getMessage()
        
        # 应用日志的自定义逻辑，同时增加类型校验避免意外报错
        msg = str(self.msg)
        if self.args:
            msg = msg % self.args
            if self.args and isinstance(self.args[0], str):
                self.session_id = self.args[0]
        return msg

logging.setLogRecordFactory(LogRecord)

方式二：为应用日志器单独添加过滤器

不全局替换LogRecordFactory，而是给应用专属日志器绑定过滤器，仅在应用日志中注入session_id：

import logging

class SessionIDFilter(logging.Filter):
    def filter(self, record):
        # 仅处理应用日志（需提前定义应用日志器名称，比如"my_dash_app"）
        if not record.name.startswith("my_dash_app"):
            return True
        
        # 从参数中提取session_id，增加容错判断
        if hasattr(record, 'args') and record.args and isinstance(record.args[0], str):
            record.session_id = record.args[0]
        else:
            record.session_id = None
        return True

# 配置应用专属日志器
app_logger = logging.getLogger("my_dash_app")
app_logger.addFilter(SessionIDFilter())

# 应用中使用该日志器记录日志
# app_logger.info("User performed action: %s", session_id)

方案对比 #

• 方式一：改动量小，适配已有全局日志的代码结构，只需在原自定义类中增加分支判断。
• 方式二：解耦性更强，不会干扰其他日志系统，适合新应用或日志架构清晰的场景。

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