如何在asyncio任务间传递contextvars.Context且无需修改任务创建逻辑

你的猜测完全正确，asyncio 任务的上下文绑定发生在任务创建的瞬间，Context.run 只是让传入的同步函数在指定上下文内执行，你之前的写法相当于仅在自定义上下文里执行了「创建任务」这个操作，任务本身捕获的仍然是创建时线程的活跃上下文，因此无法生效。

方案1（Python 3.11+ 原生支持，无侵入） #

Python 3.11 为 asyncio.create_task 新增了 context 参数，专门用于指定任务要继承的上下文，不需要重构任何原有业务逻辑，示例代码如下：

import asyncio
import contextvars

ctxvar = contextvars.ContextVar("ctx")

async def lifespan():
    ctxvar.set("spam")
    print("set")
    # 执行结束返回当前上下文快照
    return contextvars.copy_context()

async def endpoint():
    print("get")
    assert ctxvar.get() == "spam"

async def main():
    lifespan_task = asyncio.create_task(lifespan())
    # 拿到lifespan执行完成后的上下文
    lifespan_final_ctx = await lifespan_task
    # 创建endpoint任务时指定继承的上下文
    endpoint_task = asyncio.create_task(endpoint(), context=lifespan_final_ctx)
    await endpoint_task

asyncio.run(main())

该方案下endpoint完全继承lifespan的所有contextvar设置，且endpoint后续对contextvar的修改不会反向影响原上下文，完全符合你的需求。

方案2（Python 3.10及以下兼容） #

如果使用的Python版本低于3.11，可以写一个简单的包装函数实现相同效果：

import asyncio
import contextvars
from typing import Coroutine, Any

def create_task_with_context(coro: Coroutine[Any, Any, Any], context: contextvars.Context) -> asyncio.Task:
    async def wrapper():
        # 驱动协程在指定上下文内运行
        result = None
        while True:
            try:
                value = context.run(coro.send, result)
            except StopIteration as e:
                return e.value
            result = await value
    return asyncio.create_task(wrapper())

# 后续用法和方案1一致，仅需要把对应位置的create_task替换为create_task_with_context即可

内容的提问来源于stack exchange，提问作者LoveToCode