如何在Django ORM查询结果中添加重复行标记列？

原查询代码 #

ws = Ws.objects.all()
ws = ws.filter(*q_objects, **filter_kwargs)
ws = ws.order_by(*sort_list) 
ws = ws.annotate(created_unix=UnixTimestamp(F('created')), date_unix=UnixTimestamp(F('date')))
total_sum = ws.aggregate(total=Sum('ts_time', output_field=DecimalField()))['total']
LIST_FIELDS[1], LIST_FIELDS[2] = 'created_unix', 'date_unix'
return JsonResponse({'table_data': list(ws.values_list(*LIST_FIELDS)), 'summ_time': total_sum}, safe=False)

需求说明 #

需要为查询结果添加一个布尔列，当某行存在其他行与它指定的多个字段（例如LIST_FIELDS[0]和LIST_FIELDS[2]对应的模型字段）完全匹配时，该列值为True，否则为False。示例效果如下：

之前的无效方案 #

此前得到的方案仅实现了单行内字段比较，无法满足跨多行匹配的需求：

ws = ws.annotate(
    match=Case(
        When(field1=F('field2'), then=Value(True)),
        default=Value(False),
        output_field=BooleanField()
    )
)

正确解决方案 #

方案一：使用Exists子查询（高效推荐） #

通过子查询检查是否存在其他行满足字段匹配条件，排除自身后返回布尔结果：

from django.db.models import Exists, OuterRef

# 替换为你实际需要匹配的模型字段（对应LIST_FIELDS[0]和LIST_FIELDS[2]的字段名）
match_fields = ['field_a', 'field_c']

ws = ws.annotate(
    has_duplicate=Exists(
        Ws.objects.filter(
            **{field: OuterRef(field) for field in match_fields},
            id__ne=OuterRef('id')  # 排除当前行自身
        )
    )
)

方案二：使用Count分组统计 #

先按目标字段分组统计数量，再判断数量是否大于1：

from django.db.models import Count, Case, When, BooleanField, Q

match_fields = ['field_a', 'field_c']

ws = ws.annotate(
    duplicate_count=Count(
        'id',
        filter=Q(**{field: F(field) for field in match_fields}),
        distinct=False
    )
).annotate(
    has_duplicate=Case(
        When(duplicate_count__gt=1, then=Value(True)),
        default=Value(False),
        output_field=BooleanField()
    )
)

后续处理 #

将新生成的has_duplicate字段加入LIST_FIELDS：

LIST_FIELDS.append('has_duplicate')

这样返回的table_data就会包含这个标记重复的布尔列。

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