如何在Django自定义日志格式化器中访问request对象？

问题描述 #

我已经实现了自定义JSON日志格式化器，将日志数据以JSON格式写入文本文件，代码如下：

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
      super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
      log_record['timestamp'] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
      log_record['level'] = record.levelname
      log_record['location'] = record.name

我希望在这个格式化器中自动访问request对象，把“唯一请求标识符”添加到日志里，以此区分不同请求对应的日志行，期望实现类似下面的效果：

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
      super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
      log_record['timestamp'] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
      log_record['level'] = record.levelname
      log_record['location'] = record.name

      log_record['request_id'] = request.uuid   # 类似这样的实现

请问有没有办法做到这一点，不用手动把request对象传递到每一条日志中？

解决方案 #

可以利用**线程本地存储（Thread Local Storage）**来实现，因为Web框架中每个请求通常由单独线程处理，线程本地存储能让同一线程内的代码共享请求信息，无需显式传递。

步骤1：创建线程本地存储对象 #

定义全局的线程本地存储实例，用于存储当前请求的唯一标识：

import threading

local_storage = threading.local()

步骤2：在请求中间件中存入request_id #

根据你使用的Web框架，在请求入口处生成或获取请求唯一标识，并存入线程本地存储：

Flask示例：

from flask import request
import uuid

@app.before_request
def set_request_id():
    # 优先取请求头中的X-Request-ID，没有则生成UUID
    request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
    local_storage.request_id = request_id

Django示例：

import uuid
from django.utils.deprecation import MiddlewareMixin

class RequestIDMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request_id = request.META.get('HTTP_X_REQUEST_ID', str(uuid.uuid4()))
        local_storage.request_id = request_id

步骤3：修改格式化器获取request_id #

更新自定义格式化器，从线程本地存储中取出request_id并添加到日志记录：

class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
      super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
      log_record['timestamp'] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
      log_record['level'] = record.levelname
      log_record['location'] = record.name

      # 从线程本地存储获取request_id，无值时设为默认标识
      log_record['request_id'] = getattr(local_storage, 'request_id', 'unknown_request')

额外注意事项 #

• 清理线程本地存储：请求结束后要删除存储的request_id，避免线程复用导致的脏数据。比如在Flask中：

  @app.after_request
  def clear_request_id(response):
      if hasattr(local_storage, 'request_id'):
          del local_storage.request_id
      return response
  

• 异步框架适配：如果使用FastAPI这类异步框架，线程本地存储不适用，需要改用contextvars模块：

  import contextvars
  from fastapi import Request
  import uuid
  
  # 定义上下文变量
  request_id_var = contextvars.ContextVar('request_id', default='unknown_request')
  
  # 在依赖中设置request_id
  async def get_request_id(request: Request):
      request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
      request_id_var.set(request_id)
      return request_id
  
  # 格式化器中获取
  class CustomJsonFormatter(jsonlogger.JsonFormatter):
      def add_fields(self, log_record, record, message_dict):
          super().add_fields(log_record, record, message_dict)
          # ...其他字段设置
          log_record['request_id'] = request_id_var.get()
  

内容的提问来源于stack exchange，提问作者Alvaro Bataller