Python logging.Filter 注入请求上下文的做法

如果每次记录日志都手动写入 request_id，调用点很快会变得重复，而且异步任务切换后还容易拿错请求。更稳妥的做法是：用 contextvars.ContextVar 保存当前请求上下文，再让 logging.Filter 在输出前把字段注入 LogRecord。这样业务代码只写 logger.info("开始处理")，格式化器仍能输出统一的请求标识。

官方文档：https://docs.python.org/3/library/logging.html

先把请求上下文放进 ContextVar

请求标识适合放在模块级的 ContextVar 中，而不是放在普通全局变量里。普通全局变量会被并发请求覆盖；ContextVar 会随当前线程或异步上下文保存不同值。下面的示例把用户标识一起保存，未进入请求边界时返回 -。

import contextvars

# ContextVar 应在模块级创建，避免每次请求都创建新的变量对象。
request_id_var = contextvars.ContextVar("request_id", default="-")
user_id_var = contextvars.ContextVar("user_id", default="-")

def enter_request(request_id: str, user_id: str):
    # set 返回 token，后续必须用同一个 token 恢复进入前的值。
    request_token = request_id_var.set(request_id)
    user_token = user_id_var.set(user_id)
    return request_token, user_token

让 logging.Filter 负责补齐 LogRecord 字段

Filter.filter 会看到经过它所在 logger 或 Handler 的记录。这里不在业务调用点拼接字符串，而是直接添加结构化字段，Formatter 只负责展示。字段总要有值，否则某条后台日志没有上下文时，格式化可能因缺少键而报错。

import logging

class RequestContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # 读取当前上下文；缺省值保证启动日志也能正常格式化。
        record.request_id = request_id_var.get()
        record.user_id = user_id_var.get()
        # 返回 True 表示不拦截这条日志，只增加上下文字段。
        return True

handler = logging.StreamHandler()
handler.addFilter(RequestContextFilter())
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s request=%(request_id)s user=%(user_id)s %(message)s"
))

logger = logging.getLogger("service")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

如果程序有多个输出目标，要明确字段的作用范围。Filter 加在某个 Handler 上，只影响这个 Handler 的 LogRecord；加在 logger 上则可能影响下游 Handler。Python 3.12 以后，Filter 还可以返回一个新的 LogRecord，适合不同 Handler 需要不同上下文副本的场景。

在请求边界设置值，并在 finally 中恢复

设置上下文不能只做一半。线程池会复用线程，异步任务也会沿用当前上下文；如果请求结束后不恢复，下一次没有 request_id 的日志就可能带上旧请求的值。

def handle_request(request_id: str, user_id: str) -> None:
    request_token, user_token = enter_request(request_id, user_id)
    try:
        logger.info("开始处理请求")
        # 业务函数内部不再重复传 request_id。
        do_work()
    finally:
        # 无论业务成功还是抛异常，都恢复进入前的上下文。
        user_id_var.reset(user_token)
        request_id_var.reset(request_token)

def do_work() -> None:
    logger.info("执行核心操作")

真实 Web 框架中，把这段逻辑放在中间件、任务包装器或统一入口最合适。不要在每个业务函数里重复设置；也不要把 token 跨请求保存到共享容器中。

常见问题与排查清单

如果日志会被多个 Handler 同时处理，先决定上下文字段是全局一致还是按输出目标隔离。简单服务可以原地补字段；复杂服务则可在 Filter 中复制 LogRecord 后再返回，减少一个 Handler 对另一个 Handler 的副作用。

相关问题

为什么不用普通全局变量保存 request_id？

普通全局变量无法区分并发请求，后写入的请求会覆盖先写入的值。ContextVar 才能按当前上下文读取。

Filter 应该加在 logger 还是 Handler？

只想影响一个输出目标时加在 Handler；希望下游多个 Handler 都获得字段时再考虑加在 logger，并检查传播链。

请求没有 request_id 时怎么办？

给 ContextVar 设置 - 或 anonymous 等可识别占位符，避免 Formatter 因字段缺失失败。