用 contextvars 在 FastAPI 中传递 request_id 到日志与响应头

问题场景

一个接口在并发时，日志需要带上同一个 request_id，方便把一次请求的多条日志串起来；响应头也应返回该值，便于客户端排查。常见做法是使用模块级变量：

current_request_id = None

请求入口赋值，日志里读取。单请求看似正常，但并发时多个协程会互相覆盖，日志出现串号；同步线程池中也可能读到别的请求值。根因是模块级变量是进程内共享状态，不随协程或任务隔离。

contextvars.ContextVar 提供上下文局部变量。每个异步任务有独立上下文，适合保存 request_id 这类请求级数据。下面给出不依赖全局字典的最小实现。

最小实现

保存为 main.py：

import asyncio
import logging
import uuid
from contextvars import ContextVar

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

request_id_ctx: ContextVar[str] = ContextVar('request_id', default='-')

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_ctx.get()
        return True

handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    '%(asctime)s %(levelname)s request_id=%(request_id)s %(name)s %(message)s'
))
handler.addFilter(RequestIdFilter())
logger = logging.getLogger('demo')
logger.setLevel(logging.INFO)
logger.addHandler(handler)
logger.propagate = False

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        rid = request.headers.get('X-Request-ID') or uuid.uuid4().hex
        token = request_id_ctx.set(rid)
        try:
            response = await call_next(request)
            response.headers['X-Request-ID'] = rid
            return response
        finally:
            request_id_ctx.reset(token)

app = FastAPI()
app.add_middleware(RequestIdMiddleware)

@app.get('/hello')
async def hello():
    logger.info('handle hello')
    return {'ok': True}

@app.get('/slow')
async def slow():
    await asyncio.sleep(0.05)
    logger.info('handle slow')
    return {'ok': True}

关键点有三个。第一，中间件在请求进入时用 set 写入 request_id，并保存 token；在 finally 中 reset，避免影响后续请求。第二，日志过滤器从 ContextVar 读取值并写入 LogRecord.request_id，格式串再输出。第三，响应头在 call_next 返回后写入，客户端可直接看到同一个值。

运行与验证

安装依赖并启动：

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn
uvicorn main:app --port 8000

先验证单个请求：

curl -i -H 'X-Request-ID: req-a' http://127.0.0.1:8000/hello
curl -i http://127.0.0.1:8000/slow

第一条响应头应包含 X-Request-ID: req-a，日志中也出现 request_id=req-a。不传请求头时，中间件生成 UUID，响应头与日志保持一致。

再验证并发。以下脚本用标准库同时发 8 个请求，每个请求带不同 request_id：

python - <<'PY'
import concurrent.futures
import urllib.request

def call(i):
    req = urllib.request.Request(
        'http://127.0.0.1:8000/hello',
        headers={'X-Request-ID': f'req-{i}'},
    )
    with urllib.request.urlopen(req) as r:
        return r.headers.get('X-Request-ID')

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
    print(list(ex.map(call, range(8))))
PY

输出应与传入值一致；查看 uvicorn 日志，每一行 request_id 应对应各自请求，不应出现多个请求共用同一值。这里验证的是日志与响应头的一致性，不涉及性能结论。

边界与限制

1. 同步端点与自建线程：FastAPI 的同步端点可能在线程池中执行。不要假设所有运行方式和版本都会自动复制上下文；在自建线程、线程池任务中，应显式把 request_id 作为参数传入，或在任务创建时使用 contextvars.copy_context()。
2. 后台任务与流式响应：响应返回后执行的后台任务、流式生成器后续片段，可能不在原请求上下文内。需要跨阶段使用时，显式传递 rid，避免只依赖 ContextVar。
3. 日志配置范围：过滤器要挂在真正输出日志的 handler 上；第三方库若使用独立 logger/handler，需要单独配置，或统一走根 logger 的 handler。
4. 外部请求头需校验：直接信任客户端传入的 X-Request-ID 有日志注入和超长头风险。生产环境应限制长度与字符集，不合法时重新生成。
5. 不是分布式追踪替代品：本文只解决单进程内请求级日志关联。跨服务链路追踪需要专门的 trace 上下文方案，request_id 可作为辅助字段。
6. 异常路径：若异常在中间件之外未被处理，响应头可能来不及写入；需要在异常处理器或更外层 ASGI 中间件统一补充。

按以上方式，request_id 的生命周期被限制在单次请求的上下文中，日志与响应头读取同一来源，能避免并发请求日志串号。