Python contextvars 如何在异步任务间传递请求标识

在 asyncio 并发处理请求时，最稳妥的做法是用 ContextVar 保存请求级标识，而不是把一个可变字典塞进全局变量。进入请求边界时调用 set()，业务深层函数用 get() 读取，离开边界时在 finally 中用 reset(token) 恢复旧值。通过 asyncio.create_task() 创建任务时，当前上下文会被复制，因此不同任务可以各自携带自己的 request_id。

先定义带默认值的 ContextVar

我通常先把变量声明在日志或请求上下文模块中，让需要记录日志的函数只依赖一个稳定的读取入口。默认值不要省略，否则某个入口忘记设置标识时，get() 会抛出 LookupError，排查起来反而打断了原始业务。

import contextvars

# 默认值用于覆盖漏设请求标识的边界情况
request_id_var = contextvars.ContextVar("request_id", default="-")

def current_request_id() -> str:
    # 深层函数不需要层层接收 request_id 参数
    return request_id_var.get()

官方资料把 ContextVar 定义为上下文局部状态，适合异步框架使用。它解决的是“当前执行上下文是谁”，不是把数据自动写入所有线程或所有进程。

在请求边界 set 并保存 token

set() 返回的 token 记录了设置前的状态。把它和请求处理放在同一个边界里，清理逻辑就不会依赖调用方是否记得恢复。这个习惯对连接池、任务复用和测试用例尤其重要。

async def handle_request(request) -> dict:
    # 请求入口生成或读取唯一标识
    token = request_id_var.set(request.headers.get("X-Request-ID", "req-local"))
    try:
        # 深层函数、日志适配器都能读取同一请求上下文
        result = await load_profile(request.user_id)
        return {"request_id": current_request_id(), "data": result}
    finally:
        # 无论成功、异常还是取消，都恢复进入前的上下文
        request_id_var.reset(token)

这里的关键不是“把值设进去”，而是成对管理 set 与 reset。如果只 set 不 reset，当前任务后续继续执行的代码可能读到已经结束请求的标识。

创建任务时利用上下文复制

当父任务先设置了请求标识，再创建子任务，子任务会取得创建瞬间的上下文副本。任务之间不会因为随后某个任务修改自己的 ContextVar 而直接覆盖对方，这正是它比共享可变字典更适合请求日志的地方。

async def run_parallel_jobs() -> list[str]:
    # 父任务先确定本次请求的标识
    request_id_var.set("req-1001")

    # create_task 在创建时复制当前上下文
    tasks = [
        asyncio.create_task(fetch_part("users")),
        asyncio.create_task(fetch_part("orders")),
    ]
    return await asyncio.gather(*tasks)

async def fetch_part(part: str) -> str:
    # 每个子任务都能读取创建时继承的请求标识
    return f"{part}:{current_request_id()}"

如果任务要使用一个明确构造的上下文，可以这样传入：

def make_task(coro, request_id: str):
    # 复制调用点上下文，再只修改副本中的请求标识
    ctx = contextvars.copy_context()
    ctx.run(request_id_var.set, request_id)
    return asyncio.create_task(coro, context=ctx)  # 把副本交给 Task

这段代码表达的是上下文边界，不代表所有库的后台任务都会自动继承它；跨线程、跨进程或由外部队列重新消费时，需要在新的执行边界重新注入标识。

把 request_id 接入日志并排查边界

日志适配器只要在真正输出的那一刻调用 current_request_id()，就能把深层业务函数的日志和请求关联起来。下面的示例用一个简单函数表达这个位置，生产环境再接入标准库 logging 的 Filter、LoggerAdapter 或结构化日志处理器。

def log_event(event: str) -> None:
    # 在日志写出前读取当前任务的上下文
    print({"event": event, "request_id": current_request_id()})

还要注意创建时机：先 set 再 create_task，子任务才会复制目标值；先创建后 set，已经创建的任务不会因为父任务后来修改而回溯更新。对于必须人工控制的场景，使用 copy_context() 构造明确副本，比猜测调用链更容易维护。

常见问题

ContextVar 能完全替代函数参数吗？

不能。它适合请求标识、租户标识、语言环境这类隐式上下文；核心业务输入仍建议使用显式参数，避免依赖隐藏状态。

为什么不用 threading.local？

threading.local() 按线程隔离，而 asyncio 的多个任务通常共享一个线程。ContextVar 按执行上下文隔离，更贴合协程切换。

copy_context() 会深拷贝所有业务对象吗？

不会把每个值递归复制成独立对象。它复制的是上下文到变量值的绑定关系；如果值本身是可变对象，仍要避免跨任务共享并修改。

官方文档在哪里？

可以复制 Python 官方资料地址：https://docs.python.org/3/library/contextvars.html；任务 API 说明见 https://docs.python.org/3/library/asyncio-task.html。