如何在Django中拦截即将执行的SQL查询？

针对你的需求——拦截Django QuerySet的查询操作并转向外部API，有两种核心实现方向，根据场景选择即可：

1. 重写QuerySet核心方法（轻量型，针对特定模型） #

Django的QuerySet通过_fetch_all()触发实际的数据获取，filter()/order_by()等方法负责构建查询条件。你可以自定义QuerySet，重写这些方法，在执行API请求前插入定制逻辑。

示例代码 #

先定义自定义QuerySet：

from django.db.models.query import QuerySet
import requests

class APIModelQuerySet(QuerySet):
    def _fetch_all(self):
        # 拦截查询执行，先处理定制逻辑
        api_params = self._parse_query_to_api_params()
        # 调用外部API
        api_response = self._call_external_api(api_params)
        # 把API结果转成模型实例
        self._result_cache = self._convert_to_model_instances(api_response)
        self._prefetch_done = True

    def _parse_query_to_api_params(self):
        # 把Django QuerySet的查询条件转成API接受的参数格式
        params = {}
        # 处理过滤条件
        if self.query.where:
            for condition in self.query.where.children:
                # 示例：处理精确匹配，可扩展其他查询类型（gt/lt等）
                if condition.lookup == "exact":
                    params[condition.lhs.name] = condition.rhs
        # 处理排序
        if self.query.order_by:
            params["order_by"] = ",".join(self.query.order_by)
        # 处理分页（如果需要）
        if self.query.low_mark or self.query.high_mark:
            params["offset"] = self.query.low_mark
            params["limit"] = self.query.high_mark - self.query.low_mark
        return params

    def _call_external_api(self, params):
        # 替换成你的外部API地址
        response = requests.get("https://your-external-api.com/resource", params=params)
        response.raise_for_status()
        return response.json()

    def _convert_to_model_instances(self, api_data):
        instances = []
        model_cls = self.model
        for item in api_data:
            instance = model_cls(**item)
            # 标记实例并非来自本地数据库
            instance._state.adding = False
            instance._state.db = "external_api"
            instances.append(instance)
        return instances

    # 重写count/exists适配Admin需求
    def count(self):
        params = self._parse_query_to_api_params()
        params["count"] = 1
        response = self._call_external_api(params)
        return response.get("total", 0)

    def exists(self):
        return self.count() > 0

然后在模型中使用这个QuerySet：

from django.db import models

class ExternalModel(models.Model):
    id = models.IntegerField(primary_key=True)
    name = models.CharField(max_length=100)
    # 其他字段和本地模型定义一致

    objects = APIModelQuerySet.as_manager()

    class Meta:
        managed = False  # 禁止Django在本地创建表
        db_table = "external_resource"  # 仅作为标识

2. 自定义数据库后端（通用型，适配全项目模型） #

如果需要让所有指向外部API的模型都复用拦截逻辑，可以自定义Django数据库后端，拦截底层的SQL执行流程，把SQL查询转成API请求。

示例代码 #

创建external_backends.py：

from django.db.backends.base.base import BaseDatabaseWrapper
from django.db.backends.base.features import BaseDatabaseFeatures
import requests
import sqlparse

class ExternalAPIFeatures(BaseDatabaseFeatures):
    supports_transactions = False
    # 根据API特性配置其他参数

class ExternalAPIDatabaseWrapper(BaseDatabaseWrapper):
    vendor = "external_api"
    features_class = ExternalAPIFeatures

    def get_new_connection(self, conn_params):
        # 无需实际数据库连接，返回空对象即可
        return None

    def create_cursor(self, name=None):
        return ExternalAPICursor(self)

class ExternalAPICursor:
    def __init__(self, connection):
        self.connection = connection
        self._result = []

    def execute(self, sql, params=None):
        # 拦截SQL执行，解析SQL转成API参数
        parsed_sql = sqlparse.parse(sql)[0]
        api_params = self._parse_sql(parsed_sql, params)
        # 调用外部API
        response = requests.get("https://your-external-api.com/resource", params=api_params)
        response.raise_for_status()
        self._result = response.json()
        return len(self._result)

    def fetchall(self):
        return self._result

    def _parse_sql(self, parsed_sql, params):
        params = {}
        # 解析SELECT语句的字段、过滤条件、排序等
        # 示例：提取WHERE条件，可根据需求扩展
        for token in parsed_sql.tokens:
            if token.ttype is None and "WHERE" in token.value:
                # 简单解析条件，实际可使用更专业的SQL解析库
                condition = token.value.split("WHERE")[-1].strip()
                if "=" in condition:
                    key, val = condition.split("=")
                    params[key.strip()] = val.strip().strip("'\"")
        return params

然后在settings.py中配置这个数据库：

DATABASES = {
    "default": {
        "ENGINE": "your_project.external_backends.ExternalAPIDatabaseWrapper",
        # 可添加API地址、密钥等配置参数
    }
}

关键注意点 #

• 若要适配Django Admin，必须重写count()、exists()等Admin依赖的QuerySet方法，否则Admin的列表页、筛选功能会失效；
• 复杂查询（如跨表关联、聚合）需要额外处理，根据外部API的能力来扩展参数解析逻辑；
• 重写QuerySet的方式更灵活，适合单模型或少量模型的场景；自定义数据库后端则适合全项目统一对接外部API的场景。

内容的提问来源于stack exchange，提问作者Роман Хисматуллин