如何在Django Rest Framework中记录API全链路错误至CSV文件

问题场景 #

需要在DRF中实现全链路API错误记录：

• 覆盖所有错误类型，包括到达视图前的JWT验证错误等
• 记录附加信息：用户名、用户IP、请求URL、访问时间、SQL查询等
• 输出到CSV文件

尝试用中间件实现时，出现以下错误：

原英文提示：The response content must be rendered before it can be accessed
翻译：响应内容必须先渲染才能访问

已掌握视图内错误的记录方式（装饰器、自定义APIView），寻求全链路解决方案。原中间件代码如下：

class ErrorLogMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.filename = settings.ERROR_LOG_PATH
        self.fieldnames = ["Username", "IP", "User Agent", "Date/Time", "Time Interval", "Query", "Message", "URL"]

    def __call__(self, request):
        if not exists(self.filename):
            with open(self.filename, "a", newline="", encoding="utf-8") as csvfile:
                writer = DictWriter(csvfile, fieldnames=self.fieldnames)
                writer.writeheader()

        before = time()
        try:
            response = self.get_response(request)

            if response.status_code != 200:
                if hasattr(response, "render"):
                    if not response.is_rendered:
                        response.render()

                error = None
                if hasattr(response, "data"):
                    error = response.data
                else:
                    html_content = response.content.decode("utf-8")
                    error = html_content.split("</title>")[0].split("<title>")[1]

                self.submit_error(request, before, error)

            return response

        except Exception as e:
            self.submit_error(request, before, e)
            return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

    def submit_error(self, request, before, error):
        after = time()
        query = ""

        for dic in connection.queries:
            query += f"{dic['sql']}\n"

        new_error_log = {
            "Username": request.user.username,
            "IP": get_user_ip(request),
            "User Agent": request.META.get("HTTP_USER_AGENT"),
            "Date/Time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "Time Interval": after - before,
            "Query": query,
            "Message": str(error),
            "URL": request.get_full_path()
        }

        with open(self.filename, "a", newline="", encoding="utf-8") as csvfile:
            writer = DictWriter(csvfile, fieldnames=self.fieldnames)
            writer.writerow(new_error_log)

核心问题分析 #

原中间件的问题：

• 主动调用response.render()会干扰DRF的响应渲染流程，尤其是非DRF生成的响应（如Django原生错误页面）
• 全局捕获Exception会覆盖DRF自带的异常处理器，导致部分错误无法被正确解析

修正方案 #

通过DRF异常处理器+中间件配合实现全链路覆盖：

1. 用DRF异常处理器处理视图内及DRF层面的错误（包括JWT验证错误）
2. 中间件仅处理非DRF抛出的异常（如Django原生404、500错误）
3. 避免主动渲染响应，依赖框架自身的渲染流程获取内容

修正后的代码 #

1. 扩展DRF异常处理器（记录DRF层面错误）

# 项目/utils.py 或专门的error_log.py文件
import csv
import time
import os
from datetime import datetime
from django.conf import settings
from django.db import connection
from django.http import HttpRequest
from rest_framework.views import exception_handler

def get_user_ip(request: HttpRequest) -> str:
    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
    if x_forwarded_for:
        ip = x_forwarded_for.split(',')[0]
    else:
        ip = request.META.get('REMOTE_ADDR')
    return ip

def drf_error_log_handler(exc, context):
    # 先调用DRF默认异常处理器获取响应
    response = exception_handler(exc, context)
    request = context['request']
    
    if response is not None:
        # 记录DRF处理的错误
        submit_error(request, exc, response.status_code)
    
    return response

def submit_error(request: HttpRequest, error, status_code=None):
    filename = settings.ERROR_LOG_PATH
    fieldnames = [
        "Username", "IP", "User Agent", "Date/Time", 
        "Time Interval", "Query", "Message", "URL", "Status Code"
    ]
    
    # 初始化CSV文件（不存在则创建表头）
    if not os.path.exists(filename):
        with open(filename, "a", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
    
    # 计算请求耗时
    time_interval = time.time() - request._start_time if hasattr(request, '_start_time') else 0
    
    # 收集SQL查询语句
    query = ""
    for dic in connection.queries:
        query += f"{dic['sql']}\n"
    
    # 构建日志条目
    new_error_log = {
        "Username": request.user.username if request.user.is_authenticated else "Anonymous",
        "IP": get_user_ip(request),
        "User Agent": request.META.get("HTTP_USER_AGENT", "Unknown"),
        "Date/Time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "Time Interval": round(time_interval, 4),
        "Query": query,
        "Message": str(error),
        "URL": request.get_full_path(),
        "Status Code": status_code if status_code else 500
    }
    
    # 写入CSV
    with open(filename, "a", newline="", encoding="utf-8") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writerow(new_error_log)

2. 修正后的中间件（记录非DRF层面错误）

# 项目/middleware.py文件
import time
from django.http import HttpResponseServerError
from .utils import submit_error

class ErrorLogMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # 记录请求起始时间
        request._start_time = time.time()
        
        try:
            response = self.get_response(request)
            
            # 处理非DRF的错误响应（如Django原生404、500页面）
            if response.status_code >= 400 and not hasattr(response, 'data'):
                content = response.content.decode('utf-8') if hasattr(response, 'content') else str(response)
                # 提取错误信息（针对Django原生错误页面）
                if '<title>' in content:
                    error_msg = content.split("</title>")[0].split("<title>")[1]
                else:
                    error_msg = f"HTTP {response.status_code} Error"
                submit_error(request, error_msg, response.status_code)
            
            return response
        
        except Exception as e:
            # 处理未被捕获的全局异常
            submit_error(request, str(e))
            return HttpResponseServerError("Internal Server Error")

配置步骤 #

1. 在settings.py中注册DRF异常处理器：

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'your_project.utils.drf_error_log_handler',
}

2. 在settings.py中添加中间件（放在DRF中间件之后）：

MIDDLEWARE = [
    # ...其他中间件
    'rest_framework.middleware.RequestMiddleware',
    'your_project.middleware.ErrorLogMiddleware',
]

3. 在settings.py中配置错误日志路径：

ERROR_LOG_PATH = os.path.join(BASE_DIR, 'error_logs.csv')

内容的提问来源于stack exchange，提问作者Aydin Kazemi