contextvars 在异步请求链中传递追踪信息

异步请求链里，最实用的做法是把 request_id 放进 ContextVar，在入口设置一次，后续协程通过 get() 读取。它不会像普通全局变量那样让并发请求互相覆盖，也不需要给每个函数都增加一个追踪参数。官方文档：https://docs.python.org/3.14/library/contextvars.html。

一、用 ContextVar 绑定请求级追踪 ID

先把追踪变量定义在模块级，但不要把具体请求值写成默认全局状态。入口拿到请求头或网关生成的 ID 后调用 set()，下游日志函数只读取当前上下文。

import asyncio
import contextvars

request_id = contextvars.ContextVar("request_id", default="-")

def write_log(message: str) -> None:
    # 日志函数只读取当前请求上下文，不把 request_id 层层传参。
    print(f"[{request_id.get()}] {message}")

async def load_profile() -> dict:
    # await 不会清空当前 Task 的上下文。
    await asyncio.sleep(0)
    write_log("读取用户资料")
    return {"ok": True}

async def handle_request(incoming_id: str) -> dict:
    # set 返回 Token，供函数结束前恢复调用方原来的值。
    token = request_id.set(incoming_id)
    try:
        write_log("开始处理请求")
        return await load_profile()
    finally:
        # 无论正常返回还是异常退出，都恢复外层上下文。
        request_id.reset(token)

这里的关键不是“变量能被任何地方访问”，而是值绑定在当前执行上下文上。函数复用时仍应把业务数据显式传递；ContextVar 更适合 request_id、租户标识、日志关联号这类横切信息。

二、asyncio Task 会怎样继承上下文

在当前 Task 中调用 asyncio.create_task() 时，Python 会复制创建点的当前上下文。两个请求即使并发运行，也会在各自的 Task 中读取自己的 ID。下面的代码只用来说明上下文边界，输出顺序不应当被当作固定执行顺序。

async def worker(label: str) -> str:
    # 每个 Task 读取自己的 ContextVar 值。
    await asyncio.sleep(0)
    return f"{label}:{request_id.get()}"

async def main() -> None:
    # 两次 set 分别建立两个请求上下文，再创建对应 Task。
    first = asyncio.create_task(handle_request("req-A"))
    second = asyncio.create_task(handle_request("req-B"))
    await asyncio.gather(first, second)

    request_id.set("req-parent")
    child = asyncio.create_task(worker("child"))
    print(await child)  # child:req-parent

如果任务在错误的父上下文里创建，继承的也会是错误值。因此，创建点比执行体更值得检查：把任务创建放在请求作用域内，或显式指定要使用的 Context。

三、显式 Context 让后台任务的来源可控

需要把某个上下文交给后台任务时，可以先用 copy_context() 取得快照，再把它传给 create_task(context=...)。这样比依赖“当前恰好是什么值”更容易审查。

async def schedule_with_context() -> asyncio.Task:
    # 先固定当前请求的上下文快照，再创建任务。
    request_id.set("req-background")
    captured = contextvars.copy_context()
    task = asyncio.create_task(
        worker("background"),
        context=captured,
    )
    return task

async def run_background() -> str:
    task = await schedule_with_context()
    # 任务完成后读取结果；这里不假设并发调度顺序。
    return await task

copy_context() 是上下文快照，不是业务对象的深拷贝；可变对象仍要自行管理并发安全。若只是普通的请求内子任务，默认继承已经足够；只有跨作用域、延后执行或需要明确审计来源时才值得显式传递。

四、Token.reset 负责清理嵌套作用域

一次 set() 对应一个 Token。临时切换租户、模拟身份或追加日志标签时，使用 try/finally 包住业务代码，并在 finally 中 reset。不要只在“正常返回”分支恢复，因为取消和异常同样会离开作用域。

def with_temporary_request_id(value: str) -> None:
    token = request_id.set(value)
    try:
        write_log("进入临时上下文")
    finally:
        # 恢复 set 之前的值，避免嵌套调用污染外层请求。
        request_id.reset(token)

Python 3.14 还支持把 Token 作为上下文管理器使用，但为了兼容较早运行环境，工程代码仍可保留显式 try/finally。升级后是否采用新写法，应由项目支持的最低 Python 版本决定。

五、跨线程时先确认上下文传播边界

asyncio.to_thread() 会把当前上下文传播到它运行的线程，因此适合把阻塞文件或轻量同步函数放到线程中，同时保留请求 ID。直接用 threading.Thread 创建线程则不要假设会自动继承；需要时显式捕获上下文，并在目标线程中调用 ctx.run()。

def sync_work() -> str:
    # 这个同步函数运行在线程中，但只读取当前上下文。
    return request_id.get()

async def run_in_thread() -> str:
    request_id.set("req-thread")
    # asyncio 负责把当前 Context 传给 to_thread 的工作函数。
    return await asyncio.to_thread(sync_work)

排查追踪 ID 丢失时，沿着四个边界检查：变量是否使用同一个 ContextVar 实例、Task 是否在正确上下文创建、临时值是否过早 reset、线程入口是否绕过了 asyncio。这样比在每一层继续增加参数更容易收敛问题。

相关问题

ContextVar 能替代所有函数参数吗？

不能。它适合日志关联号、租户和请求范围的横切状态；业务输入、返回值和会影响核心逻辑的依赖仍应显式传递。

为什么并发任务会读到同一个 request_id？

通常是任务在同一个父上下文中创建，或把可变对象放进 ContextVar 后又被多个任务共享。检查任务创建点，并让上下文值保持不可变或按任务创建独立对象。