用 Python contextvars 实现“无感”请求上下文，告别全局变量污染！

“全局变量是万恶之源”——但如果你能安全地用它，为什么不呢？
本文带你用 contextvars 打造线程安全、协程隔离、无感透传的请求上下文系统。

在异步 Web 开发中（比如 FastAPI、Quart、Sanic），我们常常需要在请求处理过程中，将一些信息（如 trace_id、user_id、request_id）从入口一路透传到服务层、数据访问层。传统做法是把这些信息当作参数，一层层往下传——代码冗余、易出错、不优雅。

有没有一种方式，让这些上下文信息像“全局变量”一样随处可用，却又彼此隔离、互不干扰？Python 3.7+ 的 contextvars 模块，正是为此而生。

今天，我们就用 不到 30 行代码，实现一个“无感”的请求上下文代理系统，并彻底理解其原理。

一、问题场景：参数传递太累了！

想象一个典型请求流程：

1	handle_request → service_layer → dao_layer

你需要把 trace_id 和 user 从入口传到最底层。如果中间还有 5 层调用，就得写 5 次参数传递，而且任何一层漏传都会导致 bug。

你可能会想：能不能像全局变量一样直接读写？

1 2	# 千万别这么干！ global_ctx = {}

但在 异步并发 环境下，多个请求共享同一个 event loop，全局变量会被并发写入，导致 A 请求的数据被 B 请求覆盖——严重数据污染！

二、解决方案：contextvars + 代理魔法

Python 的 contextvars.ContextVar 是专为异步上下文设计的变量容器，它能保证：

每个 协程任务 拥有独立的上下文副本；
上下文在 await 调用中自动透传；
线程安全、协程安全。

但直接使用 ctx_var.get() / ctx_var.set() 依然不够优雅。我们稍作封装，让它看起来就像一个全局对象！

核心代码（仅 25 行）：

def bind_contextvar(contextvar):
    class ContextVarBind:
        __slots__ = ()
        def __getattr__(self, name):
            return getattr(contextvar.get(), name)
        def __setattr__(self, name, value):
            setattr(contextvar.get(), name, value)
        def __getitem__(self, index):
            return contextvar.get()[index]
        def __setitem__(self, index, value):
            contextvar.get()[index] = value
    return ContextVarBind()

这个 bind_contextvar 返回一个代理对象，它会自动调用 contextvar.get() 获取当前协程的上下文实例，并代理其属性和赋值操作。

定义上下文结构

class RequestCtx:
    __slots__ = ('trace_id', 'user')
    def __init__(self, trace_id, user):
        self.trace_id = trace_id
        self.user = user

创建上下文变量与代理

1 2	req_cv = contextvars.ContextVar('request') ctx_proxy = bind_contextvar(req_cv) # 这就是我们的“全局变量”！

三、业务代码：完全无感使用！

现在，你的业务逻辑可以像这样写：

async def service_layer():
    print(f'{ctx_proxy.trace_id} | hello {ctx_proxy.user}')
    ctx_proxy.user = ctx_proxy.user.upper()  # 直接赋值！
    await dao_layer()

async def dao_layer():
    print(f'dao got user={ctx_proxy.user}')

没有 .get()，没有参数传递，一切如全局变量般自然！

而在请求入口，只需初始化一次上下文：

async def handle_request(i: int):
    ctx = RequestCtx(trace_id=f'trace-{i}', user=f'user{i}')
    token = req_cv.set(ctx)   # 绑定当前协程上下文
    await service_layer()
    req_cv.reset(token)       # 清理（非必须，但推荐）

四、并发测试：三个请求互不干扰

1
2
3

async def main():
    tasks = [handle_request(i) for i in range(1, 4)]
    await asyncio.gather(*tasks)

输出示例：

[12.345678] trace-1 | hello user1
[12.345701] trace-1 | dao got user=USER1
[12.345800] trace-2 | hello user2
[12.345820] trace-2 | dao got user=USER2
[12.345900] trace-3 | hello user3
[12.345920] trace-3 | dao got user=USER3

✅ 每个请求的 trace_id 和 user 完全隔离，即使并发执行也互不污染！

五、额外技巧：手动切换上下文（高级用法）

你甚至可以在一个协程内临时切换上下文：

def demo_switch():
    c1 = RequestCtx('manual-1', 'alice')
    c2 = RequestCtx('manual-2', 'bob')

    t1 = req_cv.set(c1)
    print(ctx_proxy.user)  # alice
    t2 = req_cv.set(c2)
    print(ctx_proxy.user)  # bob
    req_cv.reset(t2)       # 回退到 c1
    print(ctx_proxy.user)  # alice
    req_cv.reset(t1)

这在单元测试、中间件插桩、调试等场景非常有用。

六、为什么这比 `threading.local` 更适合异步？

threading.local 只隔离线程，不隔离协程；
在异步框架中，多个请求在同一个线程中并发执行，threading.local 会串数据；
contextvars 是 Python 官方为 asyncio 设计的上下文隔离机制，原生支持协程。

七、实际应用场景

请求链路追踪（trace_id）
当前用户认证信息（user_id, role）
多租户上下文（tenant_id）
语言/时区偏好（locale, timezone）
事务上下文（db_session）

你可以将 ctx_proxy 封装为 g（类似 Flask 的 g 对象），在 FastAPI 中通过中间件自动注入：

@app.middleware("http")
async def inject_ctx(request, call_next):
    ctx = RequestCtx(...)
    token = req_cv.set(ctx)
    try:
        return await call_next(request)
    finally:
        req_cv.reset(token)