Python 请求级上下文数据的管理方案

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免费学习笔记（深入）”；

• 定义：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）注入字段

事情说清了就结束