Python 日志链路追踪实战：用 contextvars 自动带上 trace_id

线上排查问题时，最怕日志像散落的纸片：一条是入口日志，一条是数据库日志，一条是第三方接口日志，但看不出来它们是不是同一次请求。尤其是 Python Web 服务或异步任务里，多请求并发运行，单靠时间排序很容易误判。

一个实用办法是给每次请求生成 trace_id，并让本次请求内的所有日志自动带上它。本文用 contextvars 和 logging.Filter 实现这个能力，不依赖具体 Web 框架，方便迁移到 Flask、FastAPI、Django、Celery 或普通脚本中。

摘要

本文会先复现“日志缺少请求标识”的问题，再创建上下文变量保存 trace_id，通过 logging 过滤器把它注入日志记录，最后补充异步任务、线程池、清理时机和常见坑。目标是让每条日志都能快速回到对应请求。

适合人群

适合正在维护 Python Web 接口、异步任务、定时脚本、数据处理服务的开发者。你需要了解基础 logging 用法，以及函数调用和异常处理。

目录

1. 为什么普通日志不够用
2. 创建 trace_id 上下文变量
3. 用 Filter 自动补日志字段
4. 封装请求入口和业务函数
5. 异步场景怎么传递
6. 常见问题和生产建议
7. 完整示例

一、为什么普通日志不够用

先看一段常见日志：

INFO receive request path=/orders/1001
INFO query order table
INFO call payment service
ERROR payment service timeout

如果同一秒内有很多请求，这几行日志可能和其它请求混在一起。我们需要一个贯穿入口、业务函数和下游调用的标识，例如：

INFO trace_id=7d9a receive request path=/orders/1001
INFO trace_id=7d9a query order table
INFO trace_id=7d9a call payment service
ERROR trace_id=7d9a payment service timeout

二、创建 trace_id 上下文变量

contextvars 可以保存“当前上下文”里的值。它比全局变量更适合并发环境，因为不同请求可以拥有各自的上下文值。

# trace_context.py
from contextvars import ContextVar


trace_id_var: ContextVar[str] = ContextVar("trace_id", default="-")


def get_trace_id() -> str:
    return trace_id_var.get()

默认值用 -，表示当前代码不在请求上下文中。这样即使后台脚本或启动日志没有设置 trace_id，日志格式也不会报错。

三、用 Filter 自动补日志字段

直接在每次 logger.info() 里手写 trace_id 很容易漏。更好的方式是给 logging 配一个过滤器，在日志记录生成时统一补字段。

# log_config.py
import logging

from trace_context import get_trace_id


class TraceIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.trace_id = get_trace_id()
        return True


def setup_logging() -> None:
    fmt = "%(asctime)s %(levelname)s trace_id=%(trace_id)s %(name)s - %(message)s"
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(fmt))
    handler.addFilter(TraceIdFilter())

    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(logging.INFO)

这里有两个关键点：第一，日志格式里增加 %(trace_id)s；第二，过滤器每次都从上下文变量读取当前值。业务代码不需要关心日志字段怎么拼。

四、封装请求入口和业务函数

下面模拟一次请求处理。入口处生成 trace_id，登记到上下文变量；请求结束时用 token 恢复旧值，避免影响下一次调用。

# app.py
import logging
import uuid

from log_config import setup_logging
from trace_context import trace_id_var


logger = logging.getLogger("order")


def query_order(order_id: str) -> dict:
    logger.info("query order table, order_id=%s", order_id)
    return {"order_id": order_id, "amount": 99}


def call_payment(order: dict) -> None:
    logger.info("call payment service, amount=%s", order["amount"])


def handle_request(path: str, order_id: str) -> dict:
    trace_id = uuid.uuid4().hex[:12]
    token = trace_id_var.set(trace_id)
    try:
        logger.info("receive request path=%s", path)
        order = query_order(order_id)
        call_payment(order)
        logger.info("request finished")
        return order
    finally:
        trace_id_var.reset(token)


if __name__ == "__main__":
    setup_logging()
    handle_request("/orders/1001", "1001")

运行后，每条日志都会自动带上同一个 trace_id。业务函数只负责写正常日志，不需要把 trace_id 在参数里一层层传下去。

五、异步场景怎么传递

在 asyncio 任务中，contextvars 通常能跟随当前任务上下文传递。下面是一个简化示例：

import asyncio
import logging
import uuid

from log_config import setup_logging
from trace_context import trace_id_var


logger = logging.getLogger("async-order")


async def load_order(order_id: str) -> dict:
    await asyncio.sleep(0.1)
    logger.info("load order, order_id=%s", order_id)
    return {"order_id": order_id}


async def handle_async_request(order_id: str) -> dict:
    token = trace_id_var.set(uuid.uuid4().hex[:12])
    try:
        logger.info("async request start")
        order = await load_order(order_id)
        logger.info("async request done")
        return order
    finally:
        trace_id_var.reset(token)


if __name__ == "__main__":
    setup_logging()
    asyncio.run(handle_async_request("1002"))

如果你把工作丢到线程池、进程池或外部队列，就不能默认指望上下文自动过去。此时建议把 trace_id 明确作为任务参数传入，或者在任务入口重新设置上下文变量。

六、常见问题和生产建议

1. 忘记 reset，导致串请求

入口处 set 后，结束时一定要 reset。推荐写在 try/finally 里，确保异常路径也能恢复。

2. 日志格式里写了字段，过滤器却没挂上

如果格式里有 %(trace_id)s，但记录里没有这个属性，logging 会报格式化错误。最简单的处理是：所有 handler 都挂同一个过滤器，或者在日志配置函数里统一创建 handler。

3. trace_id 不能太长，也不能太随意

日志量很大时，过长的标识会增加存储成本。一般 8 到 16 个十六进制字符已经能满足很多内部排查场景；如果要跨服务追踪，可以接入统一链路追踪规范。

4. 不要把用户敏感信息当 trace_id

不要直接使用手机号、邮箱、身份证、完整 token 作为日志标识。日志会进入检索系统和归档系统，标识应当避免包含敏感信息。

七、完整示例

把上面的文件放在同一目录后，可以这样运行：

python app.py

你会看到类似输出：

2026-06-12 21:10:00,000 INFO trace_id=a12b34c56d78 order - receive request path=/orders/1001
2026-06-12 21:10:00,001 INFO trace_id=a12b34c56d78 order - query order table, order_id=1001
2026-06-12 21:10:00,002 INFO trace_id=a12b34c56d78 order - call payment service, amount=99
2026-06-12 21:10:00,003 INFO trace_id=a12b34c56d78 order - request finished

有了这个标识，排查时可以直接搜索 trace_id，把一次请求的完整路径拉出来看。

总结

Python 项目要做日志链路追踪，不一定一开始就引入庞大的追踪系统。先用 contextvars 保存当前请求标识，再用 logging.Filter 自动注入日志字段，就能解决大量日常排查问题。记住三点：入口设置，结束恢复，所有日志 handler 统一挂过滤器。

参考资料

本文参考 Python 官方文档中关于 logging、logging.Filter 和 contextvars 的说明，示例和排查建议为原创整理。