Django对union查询集的注解字段进行过滤的可选方案有哪些？

方案1：将过滤逻辑提前到每个子查询执行（最低改动成本） #

Django原生不支持对union后的查询集做过滤，你可以把过滤条件提前放到每个生成的子查询中，再执行union操作。如果需要动态支持不同的过滤条件，可以给函数加参数适配：

def annotated_lessons_queryset(filter_on_check=None):
    lessons = None
    for course_assignment in CourseAssignment.objects.all():
        qs = Lesson.objects.filter(
            course=course_assignment.course
        ).annotate(
            completed_progresses=Count(
                'progresses',
                filter=Q(group=course_assignment.group),
                output_field=IntegerField()
            ),
            on_check=Exists(
                AssignmentProgress.objects.filter(
                    lesson=OuterRef('id'), group=course_assignment.group, status='on_check'
                )
            )
        )
        # 提前对子查询执行过滤
        if filter_on_check is not None:
            qs = qs.filter(on_check=filter_on_check)
        lessons = qs if lessons is None else lessons.union(qs)
    return lessons

# 调用时直接传参即可拿到过滤后的结果
qs = annotated_lessons_queryset(filter_on_check=True)

方案2：重构查询逻辑，完全去掉union（更高效的推荐方案） #

你当前的循环union本质是要获取「课时+课程分配分组」的组合对应统计值，完全可以通过一次关联查询实现，避免循环查询和union的限制，性能远高于遍历union的实现：

from django.db.models import F, Count, Exists, OuterRef

def annotated_lessons_queryset():
    # 关联CourseAssignment直接拿到所有课时+对应分组的组合
    return Lesson.objects.filter(
        course__courseassignment__isnull=False
    ).annotate(
        # 关联获取对应的分组ID
        assignment_group_id=F('course__courseassignment__group_id'),
        # 按关联的分组统计完成作业数
        completed_progresses=Count(
            'progresses',
            filter=Q(group_id=F('assignment_group_id')),
            output_field=IntegerField()
        ),
        # 按关联的分组判断是否存在审核中作业
        on_check=Exists(
            AssignmentProgress.objects.filter(
                lesson=OuterRef('id'), 
                group_id=OuterRef('assignment_group_id'), 
                status='on_check'
            )
        )
    )

# 重构后支持任意后续过滤、排序操作
qs = annotated_lessons_queryset().filter(on_check=True)

方案3：将union结果包装为子查询后过滤 #

如果必须保留原有union逻辑，可以把union后的查询集作为子查询，再对其进行过滤：

from django.db.models import Subquery, IntegerField, BooleanField

def annotated_lessons_queryset():
    lessons = None
    for course_assignment in CourseAssignment.objects.all():
        qs = Lesson.objects.filter(
            course=course_assignment.course
        ).annotate(
            completed_progresses=Count(
                'progresses',
                filter=Q(group=course_assignment.group),
                output_field=IntegerField()
            ),
            on_check=Exists(
                AssignmentProgress.objects.filter(
                    lesson=OuterRef('id'), group=course_assignment.group, status='on_check'
                )
            )
        ).values('id', 'completed_progresses', 'on_check')
        lessons = qs if lessons is None else lessons.union(qs)
    
    # 把子查询包装后再关联主表查询，支持过滤操作
    return Lesson.objects.filter(
        id__in=Subquery(lessons.values('id'))
    ).annotate(
        completed_progresses=Subquery(
            lessons.filter(id=OuterRef('id')).values('completed_progresses')[:1],
            output_field=IntegerField()
        ),
        on_check=Subquery(
            lessons.filter(id=OuterRef('id')).values('on_check')[:1],
            output_field=BooleanField()
        )
    )

# 可以正常执行过滤
qs = annotated_lessons_queryset().filter(on_check=True)

内容的提问来源于stack exchange，提问作者Михаил Павлов