Python logging 怎样用 contextvars 注入请求标识

在异步服务里，最实用的做法是把请求标识放进 ContextVar，在请求入口调用 set()，再让日志处理器上的 Filter 从当前上下文取值并写入 LogRecord。这样业务函数不用层层传递 request_id，同一事件循环中的并发任务也不会因为共享全局变量而串号。

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

下面的示例使用标准库完成这件事，关键是三点：请求开始时保存 set() 返回的 token，请求结束时在 finally 中恢复；把注入逻辑放到 Handler 的过滤器上；为请求外日志准备一个安全的默认值。

图1：请求标识从请求边界进入上下文，再由日志过滤器写入记录字段的原创结构说明图。

先把请求范围和日志记录分开

ContextVar 保存的是当前上下文中的值，不是一个供所有协程读写的普通全局变量。Python 文档说明它适合上下文局部状态，并且原生支持 asyncio。因此可以让请求入口负责设置值，业务层只负责记录事件，日志层再负责把值补到记录中。

变量应该声明一次，并给出默认值。默认值不是用来掩盖调用错误，而是为了让启动日志、定时任务和异常处理日志在没有请求上下文时仍然能够被格式化。

import logging
from contextvars import ContextVar

# 请求外产生的日志使用短占位符，避免 Formatter 找不到字段。
request_id_var = ContextVar("request_id", default="-")

logger = logging.getLogger("demo")
logger.setLevel(logging.INFO)

用 Handler Filter 统一补上 request_id

过滤器不一定只做丢弃判断，也可以修改传给 Handler 的 LogRecord。把过滤器安装到 Handler 上，来自子 logger 的记录也能在输出前经过同一层注入；如果只把过滤器挂到某个父 Logger 上，子 Logger 产生的记录未必会经过那个 Logger 的过滤器。

class RequestContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # 由日志层读取当前请求上下文，不要求每个调用点传 extra。
        record.request_id = request_id_var.get()
        return True


handler = logging.StreamHandler()
# 格式化器始终能拿到 request_id，包括脱离请求的启动日志。
handler.setFormatter(logging.Formatter(
    "%(levelname)s [request_id=%(request_id)s] %(message)s"
))
handler.addFilter(RequestContextFilter())
logger.addHandler(handler)

这里的过滤器只负责添加字段，不改变日志级别，也不把请求对象本身塞进记录。记录中保留字符串形式的标识，通常更适合文本日志、集中式采集和按字段检索。

在请求入口 set，在 finally 中 reset

set() 返回的 token 代表设置前的状态。保存它并在 finally 中调用 reset()，可以保证正常返回和异常退出都恢复上下文。不要只写 set() 而省略恢复，否则复用执行资源的后续任务可能读到上一个请求的标识。

import asyncio


async def load_profile(user_id: int) -> dict:
    # 模拟会继续记录日志的下游协程。
    logger.info("加载用户资料 user_id=%s", user_id)
    await asyncio.sleep(0)
    return {"user_id": user_id}


async def handle_request(request_id: str, user_id: int) -> None:
    # token 让本次请求可以精确恢复进入前的上下文。
    token = request_id_var.set(request_id)
    try:
        logger.info("请求开始")
        await load_profile(user_id)
        logger.info("请求完成")
    except Exception:
        # 记录异常时仍在请求上下文内，便于按 request_id 检索。
        logger.exception("请求失败")
        raise
    finally:
        # 无论成功还是失败，都不能把标识留给后续工作。
        request_id_var.reset(token)


async def main() -> None:
    # 每个 Task 在创建时取得当前上下文的副本；这里分别设置请求标识。
    await asyncio.gather(
        handle_request("R-2048", 7),
        handle_request("R-2051", 8),
    )


if __name__ == "__main__":
    asyncio.run(main())

并发执行时，两条请求日志应分别带有 R-2048 和 R-2051。这里没有把 request_id 放到模块级可变字符串里，因为那种写法会让任务之间互相覆盖。

create_task 的继承和后台任务边界

asyncio 的 Task 默认会复制创建时的当前上下文，所以在请求入口设置标识后创建的子任务通常可以读到同一个值。这个行为适合一个请求内部的并发工作，例如同时查缓存和数据库。

但“能继承”不等于“生命周期自动正确”。如果创建了会在请求返回后继续运行的后台任务，就要明确它是否仍属于原请求：属于原请求时可以等待它结束；不属于原请求时应显式建立新的上下文，或者给它单独的任务标识，避免把已结束请求的标识带到长期任务里。

图2：set/reset 控制请求范围，Task A 和 Task B 各自携带标识；这是上下文关系说明图，不是运行截图。

Filter、LoggerAdapter 和 LogRecord 工厂怎么选

如果目标是让一组 Handler 的输出都带上当前请求标识，Handler Filter 通常最直接：它靠近输出边界，不要求业务函数修改调用方式。若只有少量业务入口需要固定附加字段，可以使用 LoggerAdapter；但它更像调用侧的包装，第三方库发出的日志不会自动经过你的 adapter。

全局 logging.setLogRecordFactory() 也能注入字段，不过它会影响进程内所有日志记录，适合有明确统一规范的基础设施层。只为一个服务或一个 Handler 添加 request_id 时，先用过滤器更容易控制影响范围。

常见坑点

• 不要在 Formatter 里直接假设字段一定存在；请求外日志会导致 KeyError，应在过滤器中写入默认值。
• 不要把过滤器只挂到父 Logger，然后认为所有后代记录都会自动被它修改；输出 Handler 是更稳定的注入边界。
• 不要在请求结束时只清空字符串；使用对应 token 的 reset() 才能恢复嵌套上下文。
• 不要把密码、完整 Authorization 头或身份证明等敏感信息放入 request_id；标识应短小、可检索且不可承载秘密。

小结

Python logging 搭配 contextvars 的最小配方是：用 ContextVar(default="-") 保存请求标识，在请求边界 set()，把返回的 token 放进 finally，再由 Handler Filter 将 get() 的结果写到 LogRecord。这样既保留了业务日志调用的简洁性，也能适应 asyncio 的并发上下文。

相关问题

contextvars 能替代所有全局配置吗？ 不能。它适合请求、任务或调用链范围的临时上下文，不适合承载需要跨请求共享的配置和连接池。

为什么不用 threading.local？ 线程本地变量只按线程隔离；多个 asyncio Task 可以在同一线程交替运行，ContextVar 更符合任务上下文的边界。

为什么日志里出现 request_id=-？ 说明记录产生时没有进入请求上下文，或者调用发生在 reset 之后。先判断这是不是启动任务、定时任务或真正丢失了请求边界。