FastAPI中如何转发请求头——链路追踪场景实践

一、无需显式传参共享请求上下文：用ContextVar #

Python 3.7+ 内置的contextvars模块是官方专为异步/多任务场景设计的上下文存储方案，每个异步任务（对应FastAPI的单个请求）会拥有独立的上下文副本，完全避免并行请求间的数据串扰，比显式传参或单例优雅得多。

具体实现步骤： #

• 定义全局ContextVar：用来存储需要共享的请求头（比如x-request-id）
• 添加FastAPI中间件：在请求进入时提取目标值并绑定到ContextVar
• 下游函数直接取值：forward_request无需接收额外参数，直接从ContextVar获取上下文数据

修改后的代码示例： #

import asyncio
import contextvars
import aiohttp
from fastapi import FastAPI, Request
import uvicorn
from datetime import datetime

app = FastAPI()

# 定义全局ContextVar，存储x-request-id
request_id_var = contextvars.ContextVar("x_request_id", default="default-id")

# 中间件：将请求头绑定到当前异步任务的上下文
@app.middleware("http")
async def set_request_context(request: Request, call_next):
    request_id = request.headers.get("x-request-id", "default-id")
    token = request_id_var.set(request_id)
    try:
        response = await call_next(request)
    finally:
        # 请求结束后重置上下文，避免内存泄漏
        request_id_var.reset(token)
    return response

@app.get("/source")
async def route1(value: int, request: Request):
    print(f'source {value}, request-id: {request_id_var.get()}')
    start = datetime.now().strftime("%H:%M:%S")
    await asyncio.sleep(5)
    # 调用forward_request时无需传递request-id
    resp: dict = await forward_request(value)
    end = datetime.now().strftime("%H:%M:%S")
    return {
        "start": start,
        "end": end,
        "value": value,
        "resp": resp,
        "request-id": request_id_var.get()
    }

@app.get("/destination")
async def route2(value, request: Request):
    incoming_request_id = request.headers.get("x-request-id")
    print(f'destination {value}, request-id: {incoming_request_id}')
    start = datetime.now().strftime("%H:%M:%S")
    await asyncio.sleep(5)
    end = datetime.now().strftime("%H:%M:%S")
    return {
        "start": start,
        "end": end,
        "value": value,
        "received-request-id": incoming_request_id
    }

async def forward_request(value: int) -> dict:
    # 直接从ContextVar获取当前请求的request-id
    current_request_id = request_id_var.get()
    async with aiohttp.ClientSession() as session:
        # 转发x-request-id到下游接口
        async with session.get(
            'http://127.0.0.1:5000/destination',
            params={'value': value},
            headers={"x-request-id": current_request_id}
        ) as resp:
            resp = await resp.json()
    return resp

if __name__ == "__main__":
    uvicornorn.run("main2:app", host="127.0.0.1", port=5000, log_level="info", workers=1)

二、确定forward_request的作用域（无需黑魔法） #

forward_request的作用域就是当前处理请求的异步任务上下文，这是由Python异步任务的上下文绑定机制天然保证的：

• FastAPI的每个请求都会启动一个独立的异步任务
• 中间件在任务启动时将x-request-id绑定到该任务专属的ContextVar
• 只要forward_request是在这个任务的调用链中（不管嵌套多少层），就能直接获取到绑定的上下文值
• 这种方式完全基于官方标准机制，无需解析调用栈，合规且易维护

内容的提问来源于stack exchange，提问作者RaamEE