Django ORM按wn分组条件SUM聚合查询失败求助

需求说明 #

需要将PotatoesMeasurement数据按wn字段分组，对alternaria字段按以下三个条件分别计数：

• 小于等于1的记录数（countSmaller）
• 处于20-30区间的记录数（countRange）
• 大于90的记录数（countBigger）

模型定义 #

import uuid
from datetime import datetime
from django.db import models
from django.core.validators import MinValueValidator, MaxValueValidator

class BasicEntity(models.Model):
    class Meta:
        abstract = True

    id = models.UUIDField(default=uuid.uuid4, editable=False, db_index=True, primary_key=True)
    created_on = models.DateTimeField(auto_now_add=True, editable=False, blank=True)
    updated_on = models.DateTimeField(auto_now=True, editable=False, blank=True)

class PotatoesMeasurement(BasicEntity):
    class Meta:
        db_table = "measurements_potatoes"

    colorado_potato_beetle_larvae = models.FloatField(
        default=0.0,
        validators=[MinValueValidator(0), MaxValueValidator(100)]
    )
    aphids_per_leaflet = models.IntegerField(blank=False, null=False)
    late_blight = models.FloatField(
        default=0.0,
        validators=[MinValueValidator(0), MaxValueValidator(100)]
    )
    alternaria = models.FloatField(
        default=0.0,
        validators=[MinValueValidator(0), MaxValueValidator(100)]
    )
    wn = models.IntegerField(
        default=datetime.now().isocalendar().week,
        validators=[MinValueValidator(1), MaxValueValidator(53)]
    )

可正常运行的Mongo原生聚合查询 #

db.measurements_potatoes.aggregate([
  {
    $group: {
      _id: "$wn",
      countSmaller: { $sum: { $cond: [{ $lte: ["$alternaria", 1] }, 1, 0] } },
      countRange: { $sum:   { $cond: [{ $and: [{ $gte: ["$alternaria", 20] }, { $lte: ["$alternaria", 30] }] }, 1, 0] } },
      countBigger: { $sum:  { $cond: [{ $gt: ["$alternaria", 90] }, 1, 0] } }
    }
  },
  {$sort: {_id: 1}},
]);

注：原Mongo查询中的$range用法有误，正确的区间判断需用$and结合$gte和$lte，上述为修正后的正确写法。

当前错误的Django ORM代码 #

res = (
    PotatoesMeasurement.objects.all()
    .values("wn")
    .aggregate(
        countSmaller=Sum(
            Case(When(alternaria__lt=1, then=1), default=0, output_field=IntegerField())
        ),
        countRange=Sum(
            Case(
                When(alternaria__range=[30, 50], then=1),
                default=0,
                output_field=IntegerField(),
            )
        ),
        countBigger=Sum(
            Case(When(alternaria__gt=1, then=1), default=0, output_field=IntegerField())
        ),
    )
    .order_by("wn")
)

错误信息 #

raise exe from e
djongo.exceptions.SQLDecodeError: 

        Keyword: None
        Sub SQL: None
        FAILED SQL: SELECT SUM(CASE WHEN "measurements_potatoes"."alternaria" < %(0)s THEN %(1)s ELSE %(2)s END) AS "countSmaller", SUM(CASE WHEN "measurements_potatoes"."alternaria" BETWEEN %(3)s AND %(4)s THEN %(5)s ELSE %(6)s END) AS "countRange", SUM(CASE WHEN "measurements_potatoes"."alternaria" > %(7)s THEN %(8)s ELSE %(9)s END) AS "countBigger" FROM "measurements_potatoes"
        Params: (1.0, 1, 0, 30.0, 50.0, 1, 0, 1.0, 1, 0)
        Version: 1.3.6

The above exception was the direct cause of the following exception:

失败原因分析 #

1. 分组逻辑错误：使用aggregate搭配values("wn")会返回全局聚合结果，而非按wn分组的每组统计，应使用annotate实现分组聚合。
2. 条件匹配错误：
   • countSmaller条件写成alternaria__lt=1，但需求是小于等于1，需改为alternaria__lte=1
   • countRange条件写成[30,50]，但需求是20-30区间，需改为[20,30]
   • countBigger条件写成alternaria__gt=1，但需求是大于90，需改为alternaria__gt=90
3. Djongo兼容性问题：Djongo在将Django ORM的Case/When转换为Mongo语法时可能存在兼容问题，尤其是复杂条件的转换。

解决方案 #

方案1：修正后的标准Django ORM写法 #

from django.db.models import Sum, Case, When, IntegerField

res = (
    PotatoesMeasurement.objects.all()
    .values("wn")  # 指定分组字段
    .annotate(
        # 小于等于1的计数
        countSmaller=Sum(
            Case(When(alternaria__lte=1, then=1), default=0, output_field=IntegerField())
        ),
        # 20-30区间的计数
        countRange=Sum(
            Case(
                When(alternaria__range=[20, 30], then=1),
                default=0,
                output_field=IntegerField(),
            )
        ),
        # 大于90的计数
        countBigger=Sum(
            Case(When(alternaria__gt=90, then=1), default=0, output_field=IntegerField())
        ),
    )
    .order_by("wn")  # 按wn升序排序
)

该写法符合Django ORM规范，修正了之前的逻辑错误，大部分场景下可正常运行。

方案2：直接执行原生Mongo聚合管道（绕开Djongo的SQL转换） #

如果Djongo仍出现转换错误，可直接传递Mongo原生聚合管道执行：

res = PotatoesMeasurement.objects.aggregate(
    pipeline=[
        {
            "$group": {
                "_id": "$wn",
                "countSmaller": {"$sum": {"$cond": [{"$lte": ["$alternaria", 1]}, 1, 0]}},
                "countRange": {"$sum": {"$cond": [{"$and": [{"$gte": ["$alternaria", 20]}, {"$lte": ["$alternaria", 30]}]}, 1, 0]}},
                "countBigger": {"$sum": {"$cond": [{"$gt": ["$alternaria", 90]}, 1, 0]}},
            }
        },
        {"$sort": {"_id": 1}}
    ],
    output_field=dict
)

该方法直接使用Mongo的聚合语法，避免了Djongo的SQL转换层，适合兼容性问题场景。

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