contextvars 是 python 3.7+ 中唯一能安全穿透异步/同步混合调用栈的请求级数据管理机制，必须在模块顶层定义 contextvar，通过 .set() 在中间件或依赖中绑定、.get() 统一读取，并手动处理跨协程和跨线程上下文传递。
用 contextvars 管理请求级数据最稳妥
Python 3.7+ 的 contextvars 是唯一能正确穿透异步/同步混合调用栈的机制，它不依赖线程局部存储（threading.local），也不受协程切换影响。如果你在 FastAPI、Starlette 或自研异步 Web 框架中存用户 ID、请求 ID、trace_id，必须用它。
常见错误是直接用全局变量或 threading.local —— 在 asyncio 中会跨请求泄漏数据，尤其当使用 asyncio.to_thread 或混合 sync/async 路由时，同一个变量可能被多个请求轮番写入。
-
contextvars.ContextVar必须在模块顶层定义，不能在函数内重复创建（否则每次都是新变量） - 读取前务必检查
.get()是否抛LookupError，不要假设一定有值 - 在中间件或入口处用
.set()绑定值，避免在子协程里多次set—— 同一 context 下重复set会覆盖，但嵌套 context 不会自动继承父值
FastAPI 中如何安全注入 request-scoped 变量
FastAPI 自身不提供请求上下文容器，但它在每个请求生命周期中会新建一个独立的 contextvars.Context。关键是在依赖项（Dependency）中完成绑定，并确保所有下游逻辑都通过同一 ContextVar 访问。
典型场景：你希望在日志里打上当前请求的 request_id，同时数据库操作函数也能拿到它做审计字段。不能靠参数层层透传，也不能靠中间件往 request.state 塞——后者只对当前 Request 对象有效，进不到异步任务或后台线程里。
立即学习“Python免费学习笔记（深入）”；
将产品展示、购物管理、资金管理等功能相结合，并提供了简易的操作、丰富的功能和完善的权限管理，为用户提供了一个低成本、高效率的网上商城建设方案包含PowerEasy CMS普及版，主要功能模块：文章频道、下载频道、图片频道、留言频道、采集管理、商城模块、商城日常操作模块500个订单限制（超出限制后只能查看和删除，不能进行其他处理） 无订单处理权限分配功能（只有超级管理员才能处理订单）
- 定义：
REQUEST_ID_VAR = contextvars.ContextVar("request_id", default=None) - 在依赖函数中
REQUEST_ID_VAR.set(generate_request_id())，并声明为Depends - 任何地方需要该值，统一调用
REQUEST_ID_VAR.get()，不要从request.state读 - 若需在后台任务（如
asyncio.create_task）中使用，必须显式复制 context：task = asyncio.create_task(coro(), context=contextvars.copy_context())
为什么不用 threading.local 或全局 dict
因为它们在 asyncio 下完全失效。一个 async def 函数可能在不同 event loop iteration 中被调度到不同线程（尤其启用 loop.run_in_executor 时），threading.local 的值会丢失；而全局 dict 则变成所有请求共享一份，导致 A 请求覆盖 B 请求的数据。
错误现象包括：日志中 request_id 错乱、权限校验拿到上一个用户的 token、数据库事务记录了错误的操作人。这类问题在线上低并发时几乎不复现，压测或流量突增时才爆发，极难定位。
-
threading.local在asyncio中无法隔离协程，仅隔离线程 - 全局 dict + request ID 作为 key？那得自己维护生命周期，容易漏清理、内存泄漏
- Flask 的
g或 Django 的thread_locals都是线程模型产物，迁移到 async 框架时必须重写
contextvars 性能和兼容性注意点
它本身开销极小（底层是 C 实现的哈希表查找），但滥用会导致隐式依赖难以测试。最大的实际限制是 Python 3.7+，如果你还在用 3.6，升级解释器比魔改上下文方案更省事。
另一个容易被忽略的点：contextvars 不会自动传播到 concurrent.futures.ThreadPoolExecutor 提交的任务里。哪怕你在主线程 set 了值，子线程里 get 到的仍是 default。
- 跨线程必须手动传递：
ctx = contextvars.copy_context()，然后executor.submit(lambda: ctx.run(your_func)) - 单元测试中，每次测试需重置 context：
contextvars.Context()新建一个干净上下文，或用contextvars.ContextVar.reset(token) - logging 模块不识别 contextvars，需配合
logging.LoggerAdapter或结构化日志库（如 structlog）注入字段
