别再瞎打日志了！Loguru + ContextVars 一套组合拳，轻松搞定全链路追踪

后端开发最头疼的事：线上出了 bug，日志里一堆信息，却不知道哪条日志对应哪个用户、哪个请求。
今天以“掌柜问数”项目为例，手把手教你用 Loguru + ContextVars 搭建生产级日志系统，让每个请求的日志都能串联起来。

一、为什么你的日志总是一团乱麻？

很多团队还在用 Python 自带的 logging 模块。不是说它不好，而是配置起来太麻烦：要写 Formatter、Handler、Filter，还要处理日志轮转、压缩、异步写入……折腾半天，往往还漏了关键信息——请求 ID。

更麻烦的是，现在大家都在用 async/await 异步协程。如果还用 threading.local 来存请求 ID，多个协程并发时会互相覆盖，日志里的 request_id 全乱套了。

解决方案：Loguru 负责简化日志输出和文件管理，ContextVars 负责在异步环境下安全传递请求 ID。两者配合，一行日志就能看到当前请求的唯一标识。

二、Loguru 到底好在哪？（小白也能懂）

Loguru 是一个第三方 Python 日志库，安装后直接用，几乎不需要配置。

• 开箱即用：from loguru import logger，然后 logger.info("hello")，控制台立刻输出带颜色、带时间、带行号的日志。
• 加文件输出只要一行：logger.add("file.log", rotation="500 MB", retention="10 days") – 自动轮转和删除旧日志。
• 异常信息自动带堆栈：logger.exception("出错啦") 直接打印完整异常信息。
• 完全兼容异步：加上 enqueue=True，日志写入不会阻塞你的 async 代码。

三、ContextVars 又是什么？和 threading.local 有啥区别？

threading.local 可以在同一个线程的不同函数间共享变量，但无法区分同一个线程里交替运行的多个协程。
ContextVars 是 Python 3.7 引入的协程安全的上下文变量。同一个协程任务中，无论你 await 多少次，变量都自动传递；不同协程之间互不干扰。

简单说：在 FastAPI / Sanic / 任何 asyncio 框架中，用 ContextVars 存 request_id = 万无一失。

四、动手打造完整日志系统（代码可运行）

下面以“掌柜问数”项目为例，给出一个精简但完整的日志模块。

4.1 项目结构

data-agent/ ├── app/ │ ├── core/ │ │ ├── context.py # 存放 request_id 上下文变量 │ │ └── logging.py # 日志配置 │ └── middleware/ │ └── request_id.py # 为每个请求生成 request_id └── main.py # FastAPI 入口

4.2 第一步：创建 contextvars 上下文 (context.py)

# app/core/context.py import contextvars import uuid # 定义一个协程安全的变量，默认值为 None request_id_ctx_var = contextvars.ContextVar("request_id", default=None) def set_request_id(req_id: str = None) -> str: """设置当前请求的 request_id，如果没有传入则自动生成""" if req_id is None: req_id = str(uuid.uuid4()) request_id_ctx_var.set(req_id) return req_id def get_request_id() -> str: """获取当前请求的 request_id""" return request_id_ctx_var.get()

4.3 第二步：配置 Loguru 日志 (logging.py)

# app/core/logging.py import sys from pathlib import Path from loguru import logger from app.core.context import get_request_id # 1. 定义日志格式，预留 extra["request_id"] 占位符 log_format = ( "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | " "<level>{level: <8}</level> | " "<magenta>request_id={extra[request_id]}</magenta> | " "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - " "<level>{message}</level>" ) # 2. 移除 Loguru 默认的控制台输出，准备重新配置 logger.remove() # 3. 定义 patched 函数：在每条日志生成前，从 ContextVars 中取出 request_id 填进去 def inject_request_id(record): rid = get_request_id() if rid is None: rid = "N/A" # 兜底，避免 KeyError record["extra"]["request_id"] = rid # 应用 patch logger = logger.patch(inject_request_id) # 4. 添加控制台输出（开发环境用） logger.add( sys.stdout, level="DEBUG", format=log_format, colorize=True, ) # 5. 添加文件输出（生产环境用），自动轮转和保留 log_path = Path("./logs") log_path.mkdir(exist_ok=True) logger.add( log_path / "app.log", level="INFO", format=log_format, rotation="1 day", # 每天轮转 retention="30 days", # 保留 30 天 encoding="utf-8", enqueue=True, # 异步写入，不阻塞主线程 ) if __name__ == "__main__": # 测试：手动设置 request_id 再打日志 from app.core.context import set_request_id set_request_id("test-123") logger.info("测试日志，应该包含 request_id=test-123")

4.4 第三步：FastAPI 中间件自动注入 request_id

# app/middleware/request_id.py from starlette.middleware.base import BaseHTTPMiddleware from app.core.context import set_request_id class RequestIDMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): # 每个请求进来时，自动生成并绑定一个 request_id req_id = set_request_id() response = await call_next(request) # 还可以把 request_id 加到返回头里，方便前端调试 response.headers["X-Request-ID"] = req_id return response

4.5 第四步：在 FastAPI 中启用中间件并验证

# main.py from fastapi import FastAPI from app.middleware.request_id import RequestIDMiddleware from app.core.logging import logger # 导入配置好的 logger app = FastAPI() app.add_middleware(RequestIDMiddleware) @app.get("/") async def root(): logger.info("处理根路径请求") return {"message": "Hello, 日志系统"} @app.get("/error") async def error(): logger.error("这是一个错误日志示例") raise ValueError("故意出错")

启动服务后，访问 http://localhost:8000/ 和 http://localhost:8000/error，控制台和日志文件里每一条日志都会带上不同的 request_id，同一个请求的多个日志拥有相同 ID。

五、这段代码到底解决了什么问题？

你随便找一个第三方 HTTP 客户端发两个并发请求，观察控制台输出：

2025-01-15 10:00:01.123 | INFO | request_id=abc-111 | ... - 处理根路径请求 2025-01-15 10:00:01.124 | INFO | request_id=def-222 | ... - 处理根路径请求

两个请求的 request_id 清晰区分，不会混淆。如果查询日志文件，grep "abc-111" app.log 就能拿到这个请求的完整调用链。

六、避坑指南（重要！）

1. 必须调用 logger.remove()：Loguru 默认会向 stderr 输出日志，不先移除会导致日志重复打印。
2. 一定要做 logger.patch(inject_request_id)：这样才能在运行时动态从 ContextVars 拿 request_id，而不是只在启动时绑定一次。
3. 文件输出要开 enqueue=True：在异步高并发场景，写文件 I/O 可能阻塞事件循环，开启队列写入能避免性能问题。
4. ContextVars 在 create_task 时注意继承：如果你手动创建新协程任务，需要传递 context=contextvars.copy_context()，否则子任务拿不到 request_id。FastAPI 的 BackgroundTasks 已经处理好继承，不用担心。

七、总结（记住这几点就够了）

一句话：Loguru 让你优雅地打日志，ContextVars 让你在异步世界里安心追踪请求。赶紧把这两招用起来，告别线上查日志时的“大海捞针”！