Django模型中存储用户自定义查询逻辑的方案探讨

针对你的需求，这里提供三个可落地的方案，解决引用完整性校验和SQL注入风险问题，同时支持任意字段、关联关系的AND/OR嵌套组合：

方案一：嵌套结构化模型存储（强校验） #

用递归关联的Django模型替代JSONField，把筛选条件拆解成逻辑节点和过滤项，数据库层面直接保证引用合法性。

模型定义 #

from django.db import models

class ConditionNode(models.Model):
    """逻辑节点：AND/OR，可嵌套子节点"""
    LOGIC_TYPES = (
        ('AND', '逻辑与'),
        ('OR', '逻辑或'),
    )
    logic_type = models.CharField(max_length=3, choices=LOGIC_TYPES)
    parent_node = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='child_nodes')

class FilterItem(models.Model):
    """具体过滤项：关联逻辑节点，指定字段、操作符和值/关联对象"""
    FIELD_CHOICES = (
        ('title', '习题标题'),
        ('difficulty', '难度'),
        ('tags', '关联标签'),
        # 补充Exercise的其他字段
    )
    OP_CHOICES = (
        ('exact', '等于'),
        ('contains', '包含'),
        ('gt', '大于'),
        ('lt', '小于'),
        ('in', '包含于'),
        # 补充合法的ORM查询操作符
    )
    condition_node = models.ForeignKey(ConditionNode, on_delete=models.CASCADE, related_name='filter_items')
    field = models.CharField(max_length=50, choices=FIELD_CHOICES)
    operator = models.CharField(max_length=10, choices=OP_CHOICES)
    # 普通字段值存这里
    plain_value = models.CharField(max_length=255, null=True, blank=True)
    # 标签关联直接存外键，保证引用完整性
    tag = models.ForeignKey('Tag', on_delete=models.CASCADE, null=True, blank=True)

# 现有模型不变
class Exercise(models.Model):
    title = models.CharField(max_length=200)
    difficulty = models.IntegerField()
    tags = models.ManyToManyField('Tag')

class Tag(models.Model):
    name = models.CharField(max_length=50, unique=True)

构建查询逻辑 #

递归解析模型结构，用Django ORM的Q对象组合条件，完全避免SQL注入：

from django.db.models import Q

def build_query_from_node(node):
    queries = []
    # 处理当前节点下的所有过滤项
    for item in node.filter_items.all():
        if item.field == 'tags':
            # 多对多标签的查询
            q = Q(tags=item.tag)
        else:
            # 普通字段的ORM查询
            lookup_key = f"{item.field}__{item.operator}"
            q = Q(**{lookup_key: item.plain_value})
        queries.append(q)
    
    # 递归处理子逻辑节点
    for child_node in node.child_nodes.all():
        child_q = build_query_from_node(child_node)
        queries.append(child_q)
    
    if not queries:
        return Q()
    
    # 根据逻辑类型组合查询
    if node.logic_type == 'AND':
        return Q(*queries, _connector=Q.AND)
    else:
        return Q(*queries, _connector=Q.OR)

# 使用示例：获取某个存储的筛选规则对应的习题
root_node = ConditionNode.objects.get(id=1)
target_exercises = Exercise.objects.filter(build_query_from_node(root_node))

优缺点：

• 优点：数据库层面强校验标签、字段合法性，适合需要长期保存用户自定义筛选规则的场景；ORM构建查询无注入风险。
• 缺点：模型结构较复杂，开发和维护需要处理递归逻辑。

方案二：动态解析前端条件（轻量灵活） #

如果不需要存储筛选条件，仅需处理前端传来的临时筛选请求，可直接解析前端提交的结构化JSON，先做合法性校验，再动态构建Q对象。

前端提交的条件格式示例 #

{
    "logic": "AND",
    "conditions": [
        {"field": "difficulty", "operator": "gt", "value": 3},
        {
            "logic": "OR",
            "conditions": [
                {"field": "tags", "value": "数学"},
                {"field": "tags", "value": "物理"}
            ]
        }
    ]
}

后端校验与查询构建 #

from django.db.models import Q
from rest_framework.exceptions import ValidationError
from .models import Exercise, Tag

def validate_and_build_query(condition_data):
    # 校验逻辑类型合法性
    logic = condition_data.get('logic')
    if logic not in ['AND', 'OR']:
        raise ValidationError("无效的逻辑操作符")
    
    queries = []
    conditions = condition_data.get('conditions', [])
    
    for item in conditions:
        if 'logic' in item:
            # 递归处理子条件
            child_q = validate_and_build_query(item)
            queries.append(child_q)
        else:
            # 校验过滤项的合法性
            field = item.get('field')
            operator = item.get('operator', 'exact')
            value = item.get('value')

            # 校验字段是否属于Exercise模型
            if field not in [f.name for f in Exercise._meta.get_fields()]:
                raise ValidationError(f"不存在的字段：{field}")
            
            # 校验操作符是否合法
            valid_ops = ['exact', 'contains', 'gt', 'lt', 'in']
            if operator not in valid_ops:
                raise ValidationError(f"无效的操作符：{operator}")
            
            # 处理标签字段的特殊校验
            if field == 'tags':
                try:
                    tag = Tag.objects.get(name=value)
                    q = Q(tags=tag)
                except Tag.DoesNotExist:
                    raise ValidationError(f"标签不存在：{value}")
            else:
                # 校验值的类型匹配
                field_type = Exercise._meta.get_field(field).get_internal_type()
                if field_type == 'IntegerField' and not isinstance(value, int):
                    try:
                        value = int(value)
                    except ValueError:
                        raise ValidationError(f"字段{field}需要整数类型值")
                # 构建普通字段的查询
                lookup_key = f"{field}__{operator}"
                q = Q(**{lookup_key: value})
            
            queries.append(q)
    
    if not queries:
        return Q()
    
    return Q(*queries, _connector=Q.AND if logic == 'AND' else Q.OR)

DRF视图中使用 #

from rest_framework.views import APIView
from rest_framework.response import Response
from .serializers import ExerciseSerializer

class ExerciseFilterView(APIView):
    def post(self, request):
        try:
            query_q = validate_and_build_query(request.data)
            exercises = Exercise.objects.filter(query_q)
            return Response(ExerciseSerializer(exercises, many=True).data)
        except ValidationError as e:
            return Response({"error": str(e)}, status=400)

优缺点：

• 优点：无需额外模型，开发成本低，适合临时筛选场景；完全通过ORM构建查询，无注入风险。
• 缺点：若需存储条件，需在代码层做前置校验，依赖后端逻辑保证数据合法性。

方案三：基于django-filter扩展（DRF生态整合） #

借助django-filter库，自定义支持嵌套逻辑的过滤器，和DRF的过滤生态无缝整合。

自定义嵌套过滤器 #

import django_filters
import json
from django.db.models import Q
from .models import Exercise
from .utils import validate_and_build_query  # 复用方案二中的校验函数

class NestedConditionFilter(django_filters.Filter):
    def filter(self, qs, value):
        if not value:
            return qs
        try:
            condition_data = json.loads(value)
            query_q = validate_and_build_query(condition_data)
            return qs.filter(query_q)
        except (json.JSONDecodeError, ValidationError) as e:
            raise django_filters.ValidationError(str(e))

class ExerciseFilterSet(django_filters.FilterSet):
    nested_condition = NestedConditionFilter(field_name='id')  # 占位字段，实际用自定义逻辑处理

    class Meta:
        model = Exercise
        fields = []

DRF视图中使用 #

from rest_framework import generics
from django_filters.rest_framework import DjangoFilterBackend

class ExerciseListView(generics.ListAPIView):
    queryset = Exercise.objects.all()
    serializer_class = ExerciseSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_class = ExerciseFilterSet

前端可通过GET参数传递结构化JSON：/exercises/?nested_condition={"logic":"AND",...}

优缺点：

• 优点：符合DRF的RESTful风格，和现有过滤逻辑整合方便；同样通过ORM构建查询，无注入风险。
• 缺点：嵌套条件的参数传递需要序列化JSON，前端处理稍繁琐。

通用安全提示 #

无论采用哪种方案，都必须遵循以下规则避免SQL注入：

1. 绝对不要直接拼接SQL语句，所有查询都通过Django ORM的Q对象或查询方法构建。
2. 对所有用户输入的字段名、操作符、值做严格校验，拒绝非法内容。

内容的提问来源于stack exchange，提问作者Samuele B.