Pythoncontextvars上下文变量

Python contextvars 上下文变量contextvars 模块提供上下文变量用于在异步任务中安全传递状态避免显式传递参数。Python 3.7 引入替代 threading.local 用于异步场景。1. ContextVar 基本用法---------------------------ContextVar 在每个上下文协程/任务中独立存储值。import contextvars# 声明上下文变量在模块级别定义request_id contextvars.ContextVar(request_id, default未知)def handle_request():模拟请求处理——读取当前上下文变量current request_id.get() # 获取当前上下文的值print(f处理请求: {current})return currentdef process_task(task_name: str):在上下文中设置变量token request_id.set(fREQ-{task_name}) # 设置值返回 tokentry:result handle_request()return resultfinally:request_id.reset(token) # 恢复之前的值# 不同调用互不干扰process_task(任务A) # 输出处理请求: REQ-任务Aprocess_task(任务B) # 输出处理请求: REQ-任务Bprint(request_id.get()) # 未知——已恢复为默认值2. ContextVar 在 asyncio 中的使用--------------------------------------每个协程各自拥有独立的上下文变量。import asyncioimport randomuser_context contextvars.ContextVar(user_context)async def fetch_data(user_id: int):模拟异步数据获取——每个协程独立上下文user_context.set(fuser_{user_id})await asyncio.sleep(random.uniform(0.1, 0.3))current user_context.get()print(f协程 {user_id}: 当前用户 {current})return currentasync def main():并发运行多个协程——上下文变量各自独立tasks [fetch_data(i) for i in range(1, 4)]await asyncio.gather(*tasks)# 注意ContextVar 在线程中也有效print(f主协程: {user_context.get(无用户)})# asyncio.run(main())# 输出顺序可能不同# 协程 1: 当前用户 user_1# 协程 2: 当前用户 user_2# 协程 3: 当前用户 user_3# 主协程: 无用户3. Context.run()——在指定上下文中执行------------------------------------------可以在特定上下文中运行代码块。import contextvarsctx_var contextvars.ContextVar(ctx_var)ctx_var.set(原始值)# 创建一个新上下文并设置值ctx contextvars.copy_context() # 复制当前上下文def run_in_context(value: str) - None:在新上下文中运行的函数ctx_var.set(value)print(f上下文内获取: {ctx_var.get()})# 在特定上下文中运行函数ctx.run(run_in_context, 上下文A的值)print(f原始上下文: {ctx_var.get()}) # 原始值——未受影响4. contextvars.copy_context() 高级用法------------------------------------------获取当前上下文的快照用于任务间传递。REQUEST_CTX contextvars.ContextVar(request_ctx)def process_in_context(ctx: contextvars.Context):在给定的上下文中处理def worker():print(f处理: {REQUEST_CTX.get()})ctx.run(worker)def create_request_context(request_data: dict) - contextvars.Context:为请求创建独立上下文ctx contextvars.copy_context() # 复制当前上下文ctx.run(lambda: REQUEST_CTX.set(request_data))return ctx# 模拟两个请求各自独立ctx_a create_request_context({id: 1, path: /api/a})ctx_b create_request_context({id: 2, path: /api/b})process_in_context(ctx_a) # 处理: {id: 1, path: /api/a}process_in_context(ctx_b) # 处理: {id: 2, path: /api/b}5. vs threading.local--------------------------threading.local 不适合异步场景——协程可能在线程间切换。import threadingthread_data threading.local()async def bad_async_example():threading.local() # 在线程之间共享不适用于协程# ContextVar 的优势# 1. 支持 asyncio——协程切换时自动维护# 2. 支持显式上下文传递copy_context run# 3. 类型安全——get()/set() 有明确语义# 下面的代码展示 threading.local 在线程中工作但异步中出问题def thread_worker(name: str):thread_data.name nameimport timetime.sleep(0.1)print(f线程: {thread_data.name})from threading import Threadt1 Thread(targetthread_worker, args(线程A,))t2 Thread(targetthread_worker, args(线程B,))# threading.local 在线程中安全但在 asyncio 协程中不可靠6. 在 Web 请求上下文中的应用----------------------------------ContextVar 非常适合在 Web 框架中传递请求上下文。import uuid# 全局上下文变量——每个请求独立current_request contextvars.ContextVar(current_request)class Request:模拟 Web 请求def __init__(self, method: str, path: str):self.id uuid.uuid4().hex[:8]self.method methodself.path pathself.headers {User-Agent: TestAgent}class RequestMiddleware:中间件——为每个请求设置上下文async def process(self, request: Request):token current_request.set(request)try:return await self._handle()finally:current_request.reset(token)async def _handle(self):req current_request.get()print(f处理请求 {req.id}: {req.method} {req.path})return 响应async def get_current_user() - str:从当前请求上下文获取用户无需传参req current_request.get()# 实际项目中从 req.headers 解析 tokenreturn f用户-{req.id}# 使用示例async def handle_web_request():req Request(GET, /api/users)middleware RequestMiddleware()return await middleware.process(req)7. 上下文传播-----------------在创建新任务时传播当前上下文。async def propagate_context():在子任务中传播父上下文ctx_var contextvars.ContextVar(trace_id)ctx_var.set(trace-12345)async def subtask():# 自动继承父任务的上下文print(f子任务 trace_id: {ctx_var.get()})# 使用 asyncio.create_task 时自动传播上下文task asyncio.create_task(subtask())await task总结contextvars 为 Python 异步编程提供了线程安全的上下文管理机制。核心优势在于协程间隔离、显式上下文传递Context.run以及与 asyncio 深度集成。在 Web 框架、日志追踪、请求处理等场景中广泛使用。