如何用Django ORM实现跨无关联字段的多字段子查询过滤

问题回顾 #

需要实现三个核心查询逻辑：

• 筛选出属于指定FilterGroup的FilterSpec集合
• 找出匹配该集合中任意一个FilterSpec（text1=r_text1且text2=r_text2）的Record实例
• 为返回的Record附加匹配到的FilterSpec的filter_name字段

要求兼容SQLite、MySQL/MariaDB，避免原生SQL或废弃方法。

基础实现方案（兼容所有SQL后端） #

利用Django的Subquery和OuterRef实现跨表关联，通过annotate将匹配的filter_name附加到Record实例上。如果一个Record匹配多个FilterSpec，此方案返回第一个匹配的filter_name（若需返回全部匹配值，见下文扩展方案）。

代码示例

from django.db.models import Subquery, OuterRef

# 1. 定位目标FilterGroup
target_filter_group = FilterGroup.objects.get(name="example-filter-group-17")

# 2. 定义子查询：匹配当前Record的FilterSpec的filter_name
matched_filter_subquery = FilterSpec.objects.filter(
    group=target_filter_group,
    r_text1=OuterRef('text1'),  # 引用外层Record的text1字段
    r_text2=OuterRef('text2')   # 引用外层Record的text2字段
).values('filter_name')[:1]  # 取第一个匹配的filter_name

# 3. 查询符合条件的Record，附加匹配的filter_name
results = Record.objects.annotate(
    matched_filter_name=Subquery(matched_filter_subquery)
).filter(
    matched_filter_name__isnull=False  # 只保留有匹配的Record
).select_related('group')  # 关联RecordGroup，方便获取其字段

用法说明

• 每个返回的Record实例会新增matched_filter_name属性，对应匹配到的FilterSpec的filter_name
• 如需获取RecordGroup的字段，直接通过record.group.further_data访问（因用了select_related，不会触发额外查询）

扩展方案：返回所有匹配的filter_name #

如果一个Record可能匹配多个FilterSpec，需要返回所有对应的filter_name，可根据数据库类型使用聚合函数：

MySQL 适配

使用Django内置的StringAgg聚合函数（Django 3.2+支持）：

from django.db.models import StringAgg, Subquery, OuterRef

matched_filters_subquery = FilterSpec.objects.filter(
    group=target_filter_group,
    r_text1=OuterRef('text1'),
    r_text2=OuterRef('text2')
).values('text1', 'text2').annotate(
    all_filter_names=StringAgg('filter_name', delimiter=', ')
).values('all_filter_names')[:1]

results = Record.objects.annotate(
    all_matched_filters=Subquery(matched_filters_subquery)
).filter(all_matched_filters__isnull=False).select_related('group')

SQLite 适配

SQLite原生支持GROUP_CONCAT，但Django没有内置对应聚合函数，需自定义：

from django.db.models import Aggregate, CharField, Subquery, OuterRef

# 自定义SQLite的GROUP_CONCAT聚合函数
class GroupConcat(Aggregate):
    function = 'GROUP_CONCAT'
    template = '%(function)s(%(expressions)s)'
    output_field = CharField()

# 构建子查询
matched_filters_subquery = FilterSpec.objects.filter(
    group=target_filter_group,
    r_text1=OuterRef('text1'),
    r_text2=OuterRef('text2')
).values('text1', 'text2').annotate(
    all_filter_names=GroupConcat('filter_name')
).values('all_filter_names')[:1]

# 查询Record
results = Record.objects.annotate(
    all_matched_filters=Subquery(matched_filters_subquery)
).filter(all_matched_filters__isnull=False).select_related('group')

方案优势 #

• 完全基于Django ORM实现，避免原生SQL和废弃的.extra()方法
• 基础方案兼容所有Django支持的SQL后端（SQLite、MySQL、PostgreSQL等）
• 利用Subquery和OuterRef实现的关联逻辑，与你提供的原生SQL逻辑完全一致，性能相当

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