如何在Python Django中通过中间件捕获并记录所有慢查询SQL？-Python教程-PHP中文网

Django的django.db.connection.queries在生产环境不可靠，因其仅在DEBUG=True时启用、每次请求后清空、且无执行时间字段；真正获取毫秒级耗时需通过CursorWrapper拦截execute/executemany，结合time.perf_counter()计时并记录SQL、参数、耗时、请求路径等上下文。

如何在python django中通过中间件捕获并记录所有慢查询sql？

为什么 Django 的 `django.db.connection.queries` 在生产环境不可靠

因为默认只在 DEBUG=True 时才记录查询，且每次请求结束后清空，无法跨请求统计耗时；更关键的是，它不包含执行时间字段，只有 SQL 文本和参数，根本没法判断“慢”。真正能拿到毫秒级耗时的，是 Django 内置的 django.db.backends.utils.CursorWrapper 执行路径，或者更底层的数据库适配器钩子。

用 `CursorWrapper` 包装器拦截所有查询（推荐方案）

这是最轻量、最可控的方式：不依赖第三方包，不修改配置，也不需要 patch 全局模块。核心是在中间件中临时替换当前线程的 cursor 类，让它在 execute 和 executemany 调用前后打点计时。

必须在中间件 process_request 中动态 patch 当前 connection 的 cursor 方法，不能提前 patch 类本身（否则影响其他请求）
注意区分 django.db.connections['default'] 和多数据库场景，需遍历所有注册的 alias
计时要用 time.perf_counter()，不是 time.time()，避免系统时间跳变干扰
记录内容至少包括：sql、params、duration_ms、stack（可选）、request.path

import time
from django.db import connections
from django.db.backends.utils import CursorWrapper
<p>class SlowQueryMiddleware:
def <strong>init</strong>(self, get_response):
self.get_response = get_response</p><pre class="brush:php;toolbar:false;">def __call__(self, request):
    # 对每个数据库连接都包装 cursor
    for conn in connections.all():
        original_cursor = conn.cursor

        def wrapped_cursor(*args, **kwargs):
            cursor = original_cursor(*args, **kwargs)
            return TrackedCursorWrapper(cursor, request)

        conn.cursor = wrapped_cursor

    response = self.get_response(request)

    # 恢复原始 cursor 方法（防止污染后续请求）
    for conn in connections.all():
        if hasattr(conn, '_original_cursor'):
            conn.cursor = conn._original_cursor
    return response

class TrackedCursorWrapper(CursorWrapper): def execute(self, sql, params=None): start = time.perf_counter() try: return super().execute(sql, params) finally: duration = (time.perf_counter() - start) * 1000 if duration > 500: # 慢查询阈值设为 500ms self.log_slow_query(sql, params, duration, 'execute')

def executemany(self, sql, param_list):
    start = time.perf_counter()
    try:
        return super().executemany(sql, param_list)
    finally:
        duration = (time.perf_counter() - start) * 1000
        if duration > 500:
            self.log_slow_query(sql, param_list, duration, 'executemany')

def log_slow_query(self, sql, params, duration, method):
    # 这里写入日志或发到 Sentry/ELK 等
    import logging
    logger = logging.getLogger('slow_sql')
    logger.warning(
        f"[{method}] {duration:.2f}ms | {sql[:100]} | params={params}"
    )

别直接用 `django.db.models.signals.post_save` 或 `pre_init`

这些信号只覆盖 ORM 层操作，完全捕获不到 raw SQL（比如 connection.cursor().execute(...)）、bulk_create 底层调用、或第三方库如 django-filter 生成的查询。信号机制本质是事件广播，而慢查询根源在数据库驱动执行环节——必须从执行入口拦截，而不是等 ORM 构建完再监听。