如何通过Django ORM实现含Case When的子查询统计功能？

让我们来一步步解决你的问题：

错误原因分析 #

你遇到的FieldError主要有两个原因：

1. 你的values('risk_level', 'pk')会让ORM同时按risk_level和pk分组，这意味着每个分组只会有一条数据，完全不符合你按风险等级统计总数的需求。
2. risk_level是通过Case/When生成的注解字段，Django ORM不允许直接对注解字段使用Count聚合函数，你应该统计原始模型的主键（比如id）来获取分组内的数量。

修正后的代码实现 #

下面的代码完全匹配你目标SQL的逻辑，同时解决了报错问题：

from django.db.models import Max, Case, When, Value, Count, CharField

# 实现目标SQL的ORM查询
risk_level_stats = TaskReport.objects.filter(subtask__pk=self.pk).annotate(
    # 计算每个报告对应的最高漏洞分数
    max_points=Max('vulns__severity_points')
).annotate(
    # 根据分数区间生成风险等级
    risk_level=Case(
        When(max_points__gte=8, then=Value('high')),
        When(max_points__gte=5, max_points__lte=7, then=Value('middle')),
        When(max_points__gte=2, max_points__lte=4, then=Value('low')),
        When(max_points__lte=1, then=Value('none')),
        default=Value(None),  # 处理没有漏洞的情况（对应SQL的ELSE NULL）
        output_field=CharField(),
    )
).values('risk_level').annotate(
    # 按风险等级分组，统计每个等级的报告数量
    count=Count('id')
).order_by('risk_level')

代码逻辑说明 #

1. 过滤&计算最高分数：先筛选指定子任务下的所有报告，然后计算每个报告关联漏洞的最高分数。
2. 生成风险等级：通过Case/When根据分数区间映射对应的风险等级，同时处理无漏洞的默认情况。
3. 分组统计：通过values('risk_level')指定按风险等级分组，再用Count('id')统计每个分组内的报告数量，这完全对应目标SQL中外层查询的COUNT("t"."id")逻辑。

验证生成的SQL #

这段ORM代码会生成与你目标SQL逻辑一致的SQL语句（简化版）：

SELECT 
    CASE 
        WHEN MAX("task_vuln"."severity_points") >= 8 THEN 'high' 
        WHEN (MAX("task_vuln"."severity_points") BETWEEN 5 AND 7) THEN 'middle' 
        WHEN (MAX("task_vuln"."severity_points") BETWEEN 2 AND 4) THEN 'low' 
        WHEN MAX("task_vuln"."severity_points") <= 1 THEN 'none' 
        ELSE NULL 
    END AS "risk_level", 
    COUNT("task_taskreport"."id") AS "count" 
FROM "task_taskreport" 
LEFT JOIN "task_taskreport_vuln" ON ... 
LEFT JOIN "task_vuln" ON ... 
WHERE "task_taskreport"."subtask_id" = 21 
GROUP BY "risk_level" 
ORDER BY "risk_level"

内容的提问来源于stack exchange，提问作者月牙天冲