Python logging Adapter适配python-json-logger输出动态task_id JSON日志

核心问题出在两处：

1. 原LoggerAdapter.process()方法仅把task_id拼接到了日志消息前缀，没有将值绑定到日志记录对象上，导致Formatter无法读取到动态传入的task_id
2. 自定义JsonFormatter中直接写死了TASK_ID的取值，没有从日志记录对象上读取Adapter传入的动态值

步骤1：修改LoggerAdapter实现 #

修改process方法，将动态获取到的task_id通过extra参数绑定到日志记录，同时保留原有的前缀拼接能力：

import logging
from pythonjsonlogger import jsonlogger
from datetime import datetime

NO_TASK_ID = "no_task_id"

class MAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        # 优先取日志调用时传入的动态task_id，不存在则用初始化Adapter时设置的默认值
        current_task_id = kwargs.pop('task_id', self.extra.get('task_id', NO_TASK_ID))
        # 保留原有的前缀拼接逻辑，不需要前缀可直接赋值为new_msg = msg
        new_msg = f"[{current_task_id}] {msg}"
        # 将task_id写入extra，后续会自动挂载到LogRecord实例属性上
        kwargs.setdefault("extra", {})["task_id"] = current_task_id
        return new_msg, kwargs

步骤2：修改自定义JSON格式化器 #

从LogRecord实例上读取Adapter传入的task_id字段，替换原有的硬编码值，同时优化时间字段取值逻辑（使用日志记录自带的生成时间，避免手动取时间的误差）：

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)
        # 生成标准UTC时间戳
        if not log_record.get("timestamp"):
            ts = datetime.utcfromtimestamp(record.created).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
            log_record["timestamp"] = ts
        # 统一日志级别为大写格式
        log_record["level"] = record.levelname.upper()
        # 读取LogRecord上挂载的动态task_id，无值时使用默认值
        log_record["TASK_ID"] = getattr(record, "task_id", NO_TASK_ID)

步骤3：修正日志初始化逻辑 #

初始化完成后返回Adapter实例，所有日志调用都通过Adapter发起，避免直接使用原生logger导致字段丢失：

def init_json_logger():
    logger = logging.getLogger(__name__)
    # 清除已有handler，避免重复打印日志
    logger.handlers.clear()
    log_handler = logging.StreamHandler()
    formatter = CustomJsonFormatter("%(timestamp)s %(level)s %(name)s %(message)s")
    log_handler.setFormatter(formatter)
    logger.addHandler(log_handler)
    logger.setLevel(logging.INFO)
    # 初始化Adapter时传入默认task_id
    adapter = MAdapter(logger, {"task_id": NO_TASK_ID})
    return adapter

使用示例 #

if __name__ == "__main__":
    log = init_json_logger()
    # 不传task_id，使用默认值
    log.info("start execute task")
    # 动态传入task_id，输出独立字段
    log.info("task running", task_id="task_20240501_001")
    log.error("task failed", task_id="task_20240501_001")

运行后输出的JSON日志会自动携带动态TASK_ID字段，同时日志消息前缀也会保留对应的task_id，不需要前缀直接修改Adapter里的new_msg赋值即可。

内容的提问来源于stack exchange，提问作者or.cohenadiv