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

Python 如何在 logging 中添加请求 ID(trace 追踪)  第1张

如果您在使用 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