Python asyncio中如何使用copy_context返回的上下文运行异步函数？

contextvars.Context.run()仅支持同步执行普通函数，要在保存的上下文环境中运行异步函数，可根据Python版本选择以下方案：

Python 3.11+ 推荐方案 #

Python 3.11及以上版本的asyncio.create_task()新增了context参数，可直接传入copy_context()返回的上下文对象，让异步任务在指定上下文环境中执行：

import asyncio
import contextvars

# 定义ContextVar变量
my_var = contextvars.ContextVar("my_var")

async def async_func():
    # 获取上下文变量值
    print(f"异步函数中获取的值: {my_var.get()}")

async def main():
    # 设置初始上下文值并复制
    my_var.set("初始值")
    saved_ctx = contextvars.copy_context()
    
    # 修改当前上下文，验证异步函数是否使用保存的上下文
    my_var.set("修改后的值")
    
    # 用保存的上下文创建异步任务
    task = asyncio.create_task(async_func(), context=saved_ctx)
    await task  # 输出：异步函数中获取的值: 初始值

asyncio.run(main())

Python 3.10及以下兼容方案 #

低版本Python需要手动临时切换上下文，执行完异步函数后恢复原上下文：

import asyncio
import contextvars

my_var = contextvars.ContextVar("my_var")

async def async_func():
    print(f"异步函数中获取的值: {my_var.get()}")

async def run_in_context(ctx, coro):
    # 保存当前异步任务的原有上下文
    original_ctx = contextvars.copy_context()
    try:
        # 将目标上下文的所有变量应用到当前任务
        for var, value in ctx.items():
            var.set(value)
        # 执行异步协程
        return await coro
    finally:
        # 恢复原有上下文，避免污染后续逻辑
        for var, value in original_ctx.items():
            var.set(value)

async def main():
    my_var.set("初始值")
    saved_ctx = contextvars.copy_context()
    my_var.set("修改后的值")
    
    await run_in_context(saved_ctx, async_func())  # 输出：异步函数中获取的值: 初始值

asyncio.run(main())

原理说明 #

每个异步任务都有独立的contextvars上下文，默认继承自创建它的父任务：

• Python 3.11+通过create_task的context参数直接指定任务上下文，无需手动处理；
• 低版本通过临时替换当前任务的上下文变量，执行后恢复，确保上下文不被污染。

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