FastAPI/Starlette中间件中根据中间件顺序保留Context Vars值的问题

架构老林

2026-4-13

我明白你想在FastAPI应用里用Context Vars把请求上下文信息传递到日志记录里的需求，这种方式确实是处理请求级上下文的常用方案，但中间件的执行顺序和Context Vars的生命周期很容易踩坑，我来帮你梳理下正确的实现方式。

首先，我们需要先定义一个全局的Context Var，用来存储请求上下文信息：

import contextvars
from typing import Dict, Optional

# 定义Context Var，默认值为None
request_context: contextvars.ContextVar[Optional[Dict[str, str]]] = contextvars.ContextVar("request_context", default=None)

接下来是中间件的实现，这里要特别注意中间件的执行顺序：FastAPI/Starlette的中间件是按添加顺序“包裹”请求的——先添加的中间件会在请求进入时先执行，响应返回时最后执行。为了让后续所有逻辑都能拿到上下文，设置Context Var的中间件要放在最前面。另外，一定要用token重置上下文，避免异步场景下的上下文泄漏：

from fastapi import FastAPI, Request
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

@app.middleware("http")
async def set_request_context(request: Request, call_next):
    # 收集你需要的请求上下文信息，比如请求ID、路径、客户端IP等
    ctx_data = {
        "request_id": request.headers.get("X-Request-ID", "unknown"),
        "request_path": request.url.path,
        "client_ip": request.client.host if request.client else "unknown"
    }
    # 设置Context Var并保存token，用于后续重置
    ctx_token = request_context.set(ctx_data)
    try:
        # 继续处理后续请求逻辑
        response = await call_next(request)
        return response
    finally:
        # 必须重置Context Var，防止上下文污染其他请求
        request_context.reset(ctx_token)

然后是日志过滤器，用来把Context Var里的信息附加到LogRecord对象上，这样日志格式就能直接使用这些字段：

class RequestContextLogFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # 获取当前请求的上下文
        current_ctx = request_context.get()
        if current_ctx:
            # 给LogRecord添加自定义属性
            record.request_id = current_ctx.get("request_id")
            record.request_path = current_ctx.get("request_path")
            record.client_ip = current_ctx.get("client_ip")
        else:
            # 处理非请求场景的日志，设置默认值避免报错
            record.request_id = "non-request"
            record.request_path = "non-request"
            record.client_ip = "non-request"
        return True

接下来要把过滤器绑定到你的logger上，同时配置日志格式，让上下文信息显示出来：

# 给logger添加过滤器
logger.addFilter(RequestContextLogFilter())

# 配置全局日志格式，包含上下文字段
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(request_id)s | %(request_path)s | %(client_ip)s | %(message)s"
)

最后可以在路由里测试一下效果：

@app.get("/demo")
async def demo_route():
    logger.info("这是一条带有请求上下文的日志")
    return {"status": "ok"}

这里再提醒几个容易踩的坑：

中间件顺序：如果你的应用还有其他中间件（比如CORS、认证中间件），一定要把设置Context Var的中间件放在最前面，否则前面的中间件里的日志拿不到上下文信息。
异步安全：Context Vars本身是异步安全的，但务必在finally块里重置，否则在任务复用的场景下（比如uvicorn的worker进程），可能会把前一个请求的上下文带到下一个请求里。
默认值处理：一定要处理Context Var为None的情况，比如应用启动时的初始化日志、后台任务的日志，避免抛出属性不存在的异常。

备注：内容来源于stack exchange，提问作者Christian