请求上下文（ContextVar）

说明

src/core/context.py 使用 Python 标准库 contextvars 管理请求级别的上下文信息，包括当前用户 ID、用户名、姓名和链路追踪 ID。这些变量在整个请求处理链路中自动传递，无需手动传参。

核心变量

# src/core/context.py
from contextvars import ContextVar

# 当前登录用户 ID
current_user_id: ContextVar[int] = ContextVar("current_user_id", default=0)
# 当前登录用户名
current_username: ContextVar[str] = ContextVar("current_username", default="")
# 当前登录用户姓名
current_realname: ContextVar[str] = ContextVar("current_realname", default="")

# 当前请求的链路追踪 ID
current_trace_id: ContextVar[str] = ContextVar("current_trace_id", default="")

工作原理

请求进入
   │
   ├─ authentication.py: 解析 JWT
   │   ├─ current_user_id.set(user_id)
   │   ├─ current_username.set(username)
   │   └─ current_realname.set(realname)
   ├─ trace_id.py: 生成 TraceID → current_trace_id.set(trace_id)
   │
   ├─ Handler → Service → Repository
   │   └─ 任意层可通过 current_realname.get() 获取用户姓名
   │
   └─ 请求结束: contextvars 自动清理

使用场景

1. Repository 层自动填充审计字段

# src/core/base_repository.py
class BaseRepository:
    def _uid(self) -> str:
        """获取当前登录用户姓名（从上下文变量读取）"""
        return current_realname.get()

    def create(self, obj):
        # 自动填充创建人（姓名）
        if hasattr(obj, "create_user"):
            obj.create_user = self._uid()
        # ...

    def update(self, pk, data):
        # 自动填充更新人（姓名）
        data["update_user"] = self._uid()
        # ...

2. 操作日志记录

# src/common/middleware/operation_log.py
# 操作日志通过异步落库，显式传入 realname（上下文在异步回调中可能已失效）
tornado.ioloop.IOLoop.current().spawn_callback(
    record_operation_log,
    realname=current_realname.get(),
    # ...
)

3. 日志输出中携带上下文

# src/core/logger.py
# JSON 日志自动携带 trace_id 和 user_id
{
    "timestamp": "2024-01-15T10:30:00",
    "level": "INFO",
    "trace_id": current_trace_id.get(),
    "user_id": current_user_id.get(),
    "message": "..."
}

设置时机

current_user_id / current_username / current_realname

在 src/common/middleware/authentication.py 的 login_required 装饰器中设置：

def login_required(func):
    def wrapper(self, *args, **kwargs):
        # 解析 JWT Token
        result = parse_payload(access_token)
        data = result["data"]

        # 设置上下文（userId / username / realname 均从 JWT payload 读取）
        current_user_id.set(int(data["userId"]))
        current_username.set(data.get("username", ""))
        current_realname.set(data.get("realname", ""))

        return func(self, *args, **kwargs)
    return wrapper

JWT 载荷

登录签发 JWT 时已将 userId、username、realname 写入 payload，认证中间件解析后一并写入上下文。旧 token 中可能不含 username / realname，.get() 会返回空字符串，重新登录后生效。

current_trace_id

在 src/common/middleware/trace_id.py 的 apply_trace_id 中设置：

def apply_trace_id(handler):
    trace_id = uuid.uuid4().hex[:16]
    current_trace_id.set(trace_id)
    handler.set_header("X-Trace-Id", trace_id)

生命周期

阶段	状态
请求进入	current_user_id = 0, current_username = "", current_realname = "", current_trace_id = ""
set_default_headers	current_trace_id 被设置
prepare (login_required)	current_user_id / current_username / current_realname 被设置
Handler → Service → Repository	可随时读取
on_finish	contextvars 自动清理（下一个请求重置）

与 Tornado 的区别

方面	Tornado	Tornado
实现方式	contextvars + 中间件注入	contextvars + 装饰器注入
DB 会话	中间件注入 contextvars	scoped_session 线程本地
清理时机	中间件 finally	on_finish

总结

请求上下文通过 contextvars 实现请求级别的全局变量传递。current_user_id / current_username / current_realname 在认证装饰器中设置，current_trace_id 在请求头处理中设置。Repository 层自动读取 current_realname 填充审计字段（create_user / update_user 写入操作者姓名），日志系统自动读取上下文变量输出链路信息。