可通过五种方法将请求ID注入Python日志:一、LoggerAdapter动态添加字段;二、自定义Filter结合contextvars;三、用structlog绑定上下文;四、Flask中利用g对象+Filter;五、异步场景用contextvars配合current_task。

如果您在使用 Python 的 logging 模块进行日志记录时,希望将每个请求的唯一标识(如请求 ID 或 trace ID)自动注入到每条日志中,以便在分布式系统中实现请求链路追踪,则需扩展 logging 的上下文能力。以下是实现此目标的多种方法:
一、使用 LoggerAdapter 注入 request_id
LoggerAdapter 允许在不修改原有日志调用方式的前提下,动态向 LogRecord 添加额外字段。通过构造带 request_id 的适配器实例,可使所有日志自动携带该字段。
1、定义一个支持 request_id 的 LoggerAdapter 子类或直接使用内置 LoggerAdapter。
2、在请求入口(如 Flask 的 before_request 或 Django 的 middleware)中生成唯一 request_id,例如使用 uuid.uuid4().hex。
立即学习“Python免费学习笔记(深入)”;
3、创建 LoggerAdapter 实例,传入原始 logger 和包含 request_id 的字典:adapter = logging.LoggerAdapter(logger, {'request_id': rid})。
4、后续所有日志调用均使用 adapter.info()、adapter.error() 等,而非原始 logger。
5、在 logging.Formatter 中配置格式字符串,例如 '%(asctime)s [%(request_id)s] %(levelname)s %(message)s'。
二、自定义 Filter 添加 request_id 到 LogRecord
logging.Filter 可在日志记录被处理前动态修改 LogRecord 属性。通过线程局部存储(threading.local)或上下文变量(Python 3.7+ 的 contextvars)保存当前请求的 request_id,并在 filter 方法中将其注入 record。
1、初始化一个 contextvars.ContextVar 实例用于存储 request_id:request_id_var = ContextVar('request_id', default=None)。
2、在请求开始时调用 request_id_var.set(rid),在请求结束时可选择重置。
3、定义一个继承 logging.Filter 的类,在 filter(record) 方法中执行 record.request_id = request_id_var.get()。
4、将该 Filter 实例添加到 Handler:handler.addFilter(RequestIdFilter())。
5、确保 Formatter 中包含 %(request_id)s 占位符,且该字段在 record 中存在。
三、使用 structlog 替代原生 logging
structlog 是专为结构化日志设计的第三方库,天然支持绑定上下文键值对。它可与标准 logging 后端集成,同时提供更灵活的上下文管理机制。
1、安装 structlog:pip install structlog。
2、配置 structlog 使用标准 logging 输出,例如调用 structlog.configure(wrapper_class=structlog.stdlib.BoundLogger, logger_factory=structlog.stdlib.LoggerFactory())。
3、在请求入口处获取或生成 request_id,并调用 structlog.get_logger().bind(request_id=rid) 获取绑定上下文的新 logger 实例。
4、后续所有日志均通过该绑定后的 logger 发出,每条日志自动包含 request_id 字段。
5、配置 logging.Formatter 支持结构化字段输出,或使用 structlog 提供的 JSONRenderer 直接输出结构化日志。
四、基于 werkzeug LocalProxy 或 Flask.g 的请求级绑定(Flask 专用)
在 Flask 应用中,可利用 Flask 的请求上下文对象 g 来存储 request_id,并结合自定义 Filter 实现自动注入,避免显式传递 logger 实例。
1、在 before_request 钩子中生成 request_id 并存入 flask.g:g.request_id = generate_request_id()。
2、定义一个 Filter 类,在 filter(record) 中尝试从 flask.g 获取 request_id:record.request_id = getattr(g, 'request_id', 'N/A')。
3、将该 Filter 添加至应用使用的 Handler。
4、确保日志格式中包含 %(request_id)s,且该字段在 record 中被正确设置。
5、注意:此方法依赖 Flask 请求上下文,仅适用于 Flask 应用且必须在请求上下文中执行。
五、使用 asyncio.current_task() + contextvars(异步场景)
在异步 Web 框架(如 FastAPI、Starlette)中,需使用 contextvars 而非 threading.local,因为协程可能跨线程调度,而 contextvars 提供任务隔离的上下文存储。
1、声明 contextvars.ContextVar:request_id_ctx = ContextVar('request_id', default='')。
2、在中间件中使用 await request_id_ctx.set(generate_id()) 绑定当前任务的 request_id。
3、定义 Filter 类,在 filter(record) 中调用 record.request_id = request_id_ctx.get()。
4、将 Filter 添加到 handler,并确保 formatter 包含 %(request_id)s。
5、必须确保每次请求都在独立的 async task 中运行,且中间件正确设置 contextvar。










