如何在Python Flask微服务日志中添加每请求唯一会话ID

核心思路 #

通过logging自定义过滤器注入请求唯一标识，搭配Flask请求钩子自动管理ID生命周期，全程无需修改现有logger.info/debug/error等业务日志调用逻辑，完全兼容已有代码。

具体实现步骤 #

1. 全局请求ID存储实现 #

用线程局部变量存储请求ID，保证多线程环境下ID不冲突：

import threading
import uuid
from flask import request

# 线程级上下文存储，不同请求的ID互不干扰
request_id_context = threading.local()

def get_request_id():
    """获取当前请求ID，非请求场景返回空字符串"""
    return getattr(request_id_context, 'request_id', '')

2. 编写日志过滤器 #

自动将请求ID注入到日志对象中，不需要修改原有日志打印代码：

import logging

class RequestIDFilter(logging.Filter):
    def filter(self, record):
        record.request_id = get_request_id()
        return True

3. 修改日志格式化配置 #

在原有格式化串中新增%(request_id)s占位符即可，修改后配置如下：

formatter = logging.Formatter('[%(asctime)s] - [%(threadName)s] [%(thread)d] - [%(request_id)s] - %(levelname)s in %(module)s at %(lineno)d: %(message)s')

4. 给日志处理器添加过滤器 #

将自定义过滤器绑定到你项目中所有用到的日志处理器（如文件处理器、控制台处理器）上：

# 替换为你项目实际使用的日志处理器实例
handler = logging.StreamHandler()
handler.addFilter(RequestIDFilter())
handler.setFormatter(formatter)

# 绑定到全局logger
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)

5. 配置Flask请求钩子自动管理ID #

在Flask应用初始化后添加钩子，自动生成/清理请求ID，支持和Java微服务链路对齐：

app = Flask(__name__) # 替换为你项目的Flask应用实例

@app.before_request
def init_request_id():
    # 优先从请求头取X-Request-ID（支持跨服务传递链路ID），不存在则自动生成
    request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()).replace('-', ''))
    request_id_context.request_id = request_id

@app.after_request
def clear_request_id(response):
    # 请求结束后清理线程变量，避免线程复用导致ID错乱
    if hasattr(request_id_context, 'request_id'):
        delattr(request_id_context, 'request_id')
    # 可选：把请求ID放到响应头返回，方便前端排查问题
    response.headers['X-Request-ID'] = get_request_id()
    return response

特殊场景适配 #

• 异步任务/非请求线程打印日志：只需在触发异步任务前传递当前请求ID，在子线程开头执行request_id_context.request_id = 传入的ID即可
• 手动指定单次日志的ID：打印日志时传入extra={'request_id': '自定义ID'}即可，优先级高于自动生成的ID

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