Program

前言 在实际业务中，根据 tracking_id 追查日志中一条请求的完整处理路径是一个比较常见的需求。不过 FastAPI 官方并没有提供相对应的功能，因此需要开发者自行实现。本文介绍如何基于 contextvars，为每次请求的完整流程都添加一个 tracking_id，并在日志中自动记录。 什么是 contextvars Python 在 3.7 版本的标准库中加入了一个模块 contextvars，顾名思义就是 “(Context Variables) 上下文变量”，通常用来隐式地传递一些环境信息的变量，其作用跟 threading.local() 比较相似。不过 threading.local() 是针对线程的，隔离线程之间的数据状态，而 contextvars 可以用在 asyncio 生态的异步协程中。PS: contextvars 不仅可以用在异步协程中，也可以替代 threading.local() 用在多线程函数中。 基本使用 首先编写 context.py import contextvars from typing import Optional TRACKING_ID: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar( 'tracking_id', default=None ) def get_tracking_id() -> Optional[str]: """用于依赖注入""" return TRACKING_ID.get() 编写中间件 middlewares.py，在请求头和响应头中添加 tracking_id 的信息。常见场景就是客户拿着 tracking_id 找碴。 import uuid from starlette.middleware.base import (BaseHTTPMiddleware, RequestResponseEndpoint) from starlette.requests import Request from starlette.responses import Response from context import TRACKING_ID class TrackingIDMiddleware(BaseHTTPMiddleware): async def dispatch( self, request: Request, call_next: RequestResponseEndpoint ) -> Response: tracking_id = str(uuid.uuid4()) token = TRACKING_ID.set(tracking_id) # HTTP 请求头习惯于使用 latin-1 编码 request.scope["headers"].append((b"x-request-id", tracking_id.encode("latin-1"))) try: resp = await call_next(request) finally: # 无论是否成功，每次请求结束时重置 tracking_id，避免泄露到下一次的请求中 TRACKING_ID.reset(token) # 可选, 在响应中设置跟踪 ID 头 resp.headers["X-Tracking-ID"] = tracking_id return resp 编写 handler 函数 handlers.py，测试在 handler 函数中获取 tracking_id。 import asyncio from context import TRACKING_ID async def mock_db_query(): await asyncio.sleep(1) current_id = TRACKING_ID.get() print(f"This is mock_db_query. Current tracking ID: {current_id}") await asyncio.sleep(1) 编写主函数 main.py import uvicorn from fastapi import Depends, FastAPI from fastapi.responses import PlainTextResponse from starlette.background import BackgroundTasks from context import TRACKING_ID, get_tracking_id from handlers import mock_db_query from middlewares import TrackingIDMiddleware app = FastAPI() app.add_middleware(TrackingIDMiddleware) @app.get("/qwer") async def get_qwer(): """测试上下文变量传递""" current_id = TRACKING_ID.get() print(f"This is get qwer. Current tracking ID: {current_id}") return PlainTextResponse(f"Current tracking ID: {current_id}") @app.get("/asdf") async def get_asdf(tracking_id: str = Depends(get_tracking_id)): """测试依赖注入""" print(f"This is get asdf. tracking ID: {tracking_id}") await mock_db_query() return PlainTextResponse(f"Get request, tracking ID: {tracking_id}") if __name__ == "__main__": uvicorn.run("main:app", host="127.0.0.1", port=8000, workers=4) 启动服务后用 curl 测试 api，在控制台可以看到 tracking_id 在请求中都能捕获到。 This is get qwer. Current tracking ID: 01b0153f-4877-4ca0-ac35-ed88ab406452 INFO: 127.0.0.1:55708 - "GET /qwer HTTP/1.1" 200 OK This is get asdf. tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639 This is mock_db_query. Current tracking ID: 0be61d8d-11a0-4cb6-812f-51b9bfdc2639 INFO: 127.0.0.1:55722 - "GET /asdf HTTP/1.1" 200 OK 使用 curl 的控制台输出 ...