如何在Django中用单查询实现关联模型的未完成任务统计

错误原因 #

你遇到的ProgrammingError是因为子查询返回了多行结果，但Subquery作为annotate的字段要求只能返回单个值（每个Project对应一个统计数）。之前的写法没有对符合条件的Task做聚合计数，导致子查询返回了所有符合条件的Task记录，而非总数。

正确实现代码 #

假设target_completion_count是Project模型的字段（如果是全局固定值，直接替换为具体数字即可），以下是符合原逻辑的单查询实现：

from django.db.models import Count, Exists, OuterRef, Subquery, IntegerField

def get_project_incomplete_tasks(user):
    # 子查询：判断当前用户是否已完成该任务
    user_has_completed = Completion.objects.filter(
        task=OuterRef('pk'),
        annotator=user
    )

    # 子查询：统计单个Project下符合条件的未完成任务数
    incomplete_task_count = Task.objects.filter(
        dataset__project=OuterRef('pk'),
        # 排除用户已完成的任务
        ~Exists(user_has_completed)
    ).annotate(
        # 计算任务的完成次数
        completion_count=Count('completions')
    ).filter(
        # 筛选完成次数小于目标值的任务
        completion_count__lt=OuterRef('target_completion_count')
    ).order_by().annotate(
        # 统计符合条件的任务总数
        total=Count('pk')
    ).values('total')

    # 为每个Project添加未完成任务数字段
    projects = Project.objects.annotate(
        incomplete_task_count=Subquery(
            incomplete_task_count,
            output_field=IntegerField()
        )
    ).prefetch_related("members", "metatemplate")
    
    return projects

关键逻辑说明 #

1. 排除用户已完成任务：用Exists子查询替代原exclude，避免因Task关联多个Completion导致的重复过滤问题，同时查询更高效。
2. 聚合统计：在子查询中先筛选符合条件的Task，再通过Count('pk')统计总数，确保子查询返回单个值，适配Subquery的要求。
3. 关联Project：通过OuterRef('pk')关联当前Project，实现对每个Project单独统计。

注意事项 #

如果target_completion_count不是Project的字段，而是全局参数，只需将completion_count__lt=OuterRef('target_completion_count')替换为具体数值，比如completion_count__lt=3（假设目标完成数是3）。

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