在Django ORM中计算分组查询注解字段的求和最大值

问题分析 #

你遇到的核心问题是：Django ORM不允许在同一个查询层级中直接对聚合字段再次进行聚合操作。你之前的写法试图在一次查询中连续使用annotate，但第一次annotate生成的AcSum是聚合结果，无法被同层级的另一个聚合函数（Max）直接引用。要实现需求，需要拆分两次分组聚合：先计算每个(Category, A)的AcSum，再基于这个结果按Category分组求最大值。

Django ORM 解决方案 #

假设你的模型关联关系正确（Relation关联A和B，B关联Category），可以通过两次分组聚合来实现需求：

from django.db.models import Sum, Max

# 第一步：计算每个 (Category, A) 的 Intensity 总和 AcSum
category_a_sum = Relation.objects.values(
    'b__category',  # 关联到Category
    'a'             # 关联到A
).annotate(
    ac_sum=Sum('intensity')
).order_by()  # 移除默认排序，避免不必要的分组干扰

# 第二步：按 Category 分组，计算每个 Category 内 AcSum 的最大值
max_per_category = category_a_sum.values(
    'b__category'
).annotate(
    max_ac_sum=Max('ac_sum')
).values(
    # 按需选择要返回的字段，比如Category的名称和最大值
    'b__category__name', 
    'max_ac_sum'
)

关键说明：

1. 第一次values('b__category', 'a') + annotate(ac_sum=Sum(...))会按Category和A分组，得到每个组合的Intensity总和。
2. 第二次对这个查询集再次调用values('b__category') + annotate(max_ac_sum=Max(...))，会基于第一次的结果，仅按Category分组，计算该分组内所有AcSum的最大值。
3. 最后的values()可以指定你需要返回的字段（比如Category名称、最大值），按需调整即可。

原生SQL 解决方案 #

如果ORM的方式在你的Django 2.x版本中存在兼容问题，或者你需要更灵活的控制，可以直接使用原生SQL：

SQL 查询语句：

SELECT
    c.id AS category_id,
    c.name AS category_name,
    MAX(sub.ac_sum) AS max_ac_sum
FROM (
    -- 子查询：计算每个 (Category, A) 的 Intensity 总和
    SELECT
        b.category_id,
        r.a_id,
        SUM(r.intensity) AS ac_sum
    FROM relation r
    INNER JOIN b ON r.b_id = b.id
    GROUP BY b.category_id, r.a_id
) AS sub
-- 关联Category表获取名称等信息（如果不需要可以省略这部分）
INNER JOIN category c ON sub.category_id = c.id
GROUP BY c.id, c.name;

在Django中执行原生SQL：

from django.db import connection

with connection.cursor() as cursor:
    cursor.execute("""
        SELECT
            c.id AS category_id,
            c.name AS category_name,
            MAX(sub.ac_sum) AS max_ac_sum
        FROM (
            SELECT
                b.category_id,
                r.a_id,
                SUM(r.intensity) AS ac_sum
            FROM relation r
            INNER JOIN b ON r.b_id = b.id
            GROUP BY b.category_id, r.a_id
        ) AS sub
        INNER JOIN category c ON sub.category_id = c.id
        GROUP BY c.id, c.name
    """)
    # 获取查询结果并转换为字典格式
    columns = [col[0] for col in cursor.description]
    result_list = [dict(zip(columns, row)) for row in cursor.fetchall()]

补充说明 #

• 如果你的模型字段名和示例不同（比如Intensity是大写、关联字段名不同），需要对应调整代码中的字段名称。
• 对于Django 2.x版本，上述ORM写法是完全兼容的，因为它没有使用高版本才有的特性（比如Window函数）。

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