如何在Django中嵌套使用ArrayAgg？工时单关联查询报错解决

需求 #

查询指定项目和日期的工时单，同时获取每个工时单关联的所有设备，以及每个设备对应的所有附件。原查询工时单及关联设备的代码可正常运行，但尝试在ArrayAgg的JSONObject中嵌套另一个ArrayAgg以获取设备附件时，触发FieldError。

模型定义 #

class TimeSheet(CommonModel):
    employee = models.ForeignKey(
        "employee.Employee", related_name="timesheets", on_delete=models.PROTECT
    )
    project = models.ForeignKey(
        "core.Project", related_name="employee_timesheets", on_delete=models.PROTECT
    )
    timesheet_date = models.DateField()
    clock_in_time = models.DateTimeField()
    clock_out_time = models.DateTimeField(null=True)


class TimesheetEquipment(CommonModel):
    equipment = models.ForeignKey(
        "equipment.Equipment",
        on_delete=models.PROTECT,
        related_name="timesheet_equipments",
    )
    resource_code = models.CharField(max_length=150)
    resource_name = models.CharField(max_length=511)
    rate = models.DecimalField(
        max_digits=30, decimal_places=decimal.ACCEPTED_DECIMAL_PLACE
    )
    unit = models.CharField(max_length=63, default="Hour")
    timesheet = models.ForeignKey(
        "timesheet.Timesheet",
        on_delete=models.CASCADE,
        related_name="timesheet_equipments",
    )
    equipment_attachments = models.ManyToManyField(
        "costcode.ResourceCode",
        related_name="timesheet_equipments",
        blank=True,
        through="timesheet.TimesheetEquipmentAttachment",
        help_text="Other Resources to be attached with this requirement",
    )
 

class TimesheetEquipmentAttachment(models.Model):
    timesheetequipment = models.ForeignKey(
        TimesheetEquipment,
        on_delete=models.CASCADE,
        related_name="timesheetequipmentattachments",
    )
    resource = models.ForeignKey(
        "resource_management.CostSheetResource",
        related_name="timesheetequipmentattachments",
        on_delete=models.PROTECT,
        null=True,
    )
    resource_code = models.CharField(max_length=150)
    resource_name = models.CharField(max_length=150)

正常运行的查询代码 #

from django.db.models.functions import JSONObject
from django.contrib.postgres.aggregates import ArrayAgg
from django.db.models import F

timesheets = (
            TimeSheet.objects.filter(project_id=project_id, timesheet_date=date)
            .select_related("employee")
            .prefetch_related(
                "timesheet_equipments",
                "timesheet_equipments__timesheetequipmentattachments",
            )
            .values("id")
            .annotate(
                timesheet_date=F("timesheet_date"),
                clock_in_time=F("clock_in_time"),
                clock_out_time=F("clock_out_time"),
                timesheet_equipments=ArrayAgg(
                    JSONObject(
                        equipment=F("timesheet_equipments__equipment_id"),
                        equipment_name=F(
                            "timesheet_equipments__equipment__equipment_name"
                        ),
                        resource_code=F("timesheet_equipments__resource_code"),
                        resource_name=F("timesheet_equipments__resource_name"),
                        rate=F("timesheet_equipments__rate"),
                        unit=F("timesheet_equipments__unit"),
                      ),
                    distinct=True,
                ),
            )
            .values(
                "id",
                "timesheet_date",
                "clock_in_time",
                "clock_out_time",
                "timesheet_equipments",
            )
            .order_by("-timesheet_date")
            .distinct()
        )

报错的嵌套查询代码 #

from django.db.models.functions import JSONObject
from django.contrib.postgres.aggregates import ArrayAgg
from django.db.models import F

timesheets = (
            TimeSheet.objects.filter(project_id=project_id, timesheet_date=date)
            .select_related("employee")
            .prefetch_related(
                "timesheet_equipments",
                "timesheet_equipments__timesheetequipmentattachments",
            )
            .values("id")
            .annotate(
                timesheet_date=F("timesheet_date"),
                clock_in_time=F("clock_in_time"),
                clock_out_time=F("clock_out_time"),
                timesheet_equipments=ArrayAgg(
                    JSONObject(
                        equipment=F("timesheet_equipments__equipment_id"),
                        equipment_name=F(
                            "timesheet_equipments__equipment__equipment_name"
                        ),
                        resource_code=F("timesheet_equipments__resource_code"),
                        resource_name=F("timesheet_equipments__resource_name"),
                        rate=F("timesheet_equipments__rate"),
                        unit=F("timesheet_equipments__unit"),
                      equipment_attachments=ArrayAgg(
                            JSONObject(
                                resource_code=F(
                                    "timesheet_equipments__timesheetequipmentattachments__resource_code"
                                ),
                                resource_name=F(
                                    "timesheet_equipments__timesheetequipmentattachments__resource_name"
                                ),
                                rate=F(
                                    "timesheet_equipments__timesheetequipmentattachments__rate"
                                ),
                            ),
                            distinct=True,
                        ),
                       
                    ),
                    distinct=True,
                ),
            )
            .values(
                "id",
                "timesheet_date",
                "clock_in_time",
                "clock_out_time",
                "timesheet_equipments",
            )
            .order_by("-timesheet_date")
            .distinct()
        )

错误信息（翻译后） #

FieldError: 无法计算None('JSONObject(...)'): 'JSONObject(...)' 是一个聚合函数

错误原因 #

Django ORM不允许在一个聚合函数（如ArrayAgg）内部的JSONObject中嵌套另一个聚合函数，这是因为SQL层面不支持这种嵌套聚合的语法，Django的查询解析器也无法处理这种嵌套结构。

解决方案 #

方案一：Python层面组装嵌套数据（推荐，简单直观） #

利用prefetch_related预取所有关联数据，然后在Python中手动构造需要的嵌套结构，避免数据库层面的复杂聚合嵌套。

from django.db.models import F

# 查询基础数据并预取关联
timesheets_queryset = (
    TimeSheet.objects.filter(project_id=project_id, timesheet_date=date)
    .select_related("employee")
    .prefetch_related(
        "timesheet_equipments",
        "timesheet_equipments__timesheetequipmentattachments",
    )
    .order_by("-timesheet_date")
)

# 手动构造嵌套结构
result = []
for timesheet in timesheets_queryset:
    equipment_list = []
    for equipment in timesheet.timesheet_equipments.all():
        # 收集当前设备的附件
        attachments = [
            {
                "resource_code": att.resource_code,
                "resource_name": att.resource_name,
                "rate": att.resource.rate if att.resource else None  # 根据实际模型调整
            }
            for att in equipment.timesheetequipmentattachments.all()
        ]
        equipment_data = {
            "equipment": equipment.equipment_id,
            "equipment_name": equipment.equipment.equipment_name,
            "resource_code": equipment.resource_code,
            "resource_name": equipment.resource_name,
            "rate": equipment.rate,
            "unit": equipment.unit,
            "equipment_attachments": attachments
        }
        equipment_list.append(equipment_data)
    
    timesheet_data = {
        "id": timesheet.id,
        "timesheet_date": timesheet.timesheet_date,
        "clock_in_time": timesheet.clock_in_time,
        "clock_out_time": timesheet.clock_out_time,
        "timesheet_equipments": equipment_list
    }
    result.append(timesheet_data)

方案二：数据库层面分步聚合（适合大数据量场景） #

先给TimesheetEquipment模型注解附件的聚合结果，再在TimeSheet的查询中引用这个注解，避免直接嵌套聚合。

from django.db.models import Subquery, OuterRef
from django.db.models.functions import JSONObject
from django.contrib.postgres.aggregates import ArrayAgg

# 第一步：给每个TimesheetEquipment注解附件的聚合
equipment_with_attachments = TimesheetEquipment.objects.filter(
    id=OuterRef("timesheet_equipments__id")
).annotate(
    attachments=ArrayAgg(
        JSONObject(
            resource_code=F("timesheetequipmentattachments__resource_code"),
            resource_name=F("timesheetequipmentattachments__resource_name"),
            rate=F("timesheetequipmentattachments__resource__rate")  # 根据实际模型调整
        ),
        distinct=True
    )
).values("attachments")

# 第二步：查询TimeSheet并关联设备及附件
timesheets = (
    TimeSheet.objects.filter(project_id=project_id, timesheet_date=date)
    .values("id")
    .annotate(
        timesheet_date=F("timesheet_date"),
        clock_in_time=F("clock_in_time"),
        clock_out_time=F("clock_out_time"),
        timesheet_equipments=ArrayAgg(
            JSONObject(
                equipment=F("timesheet_equipments__equipment_id"),
                equipment_name=F("timesheet_equipments__equipment__equipment_name"),
                resource_code=F("timesheet_equipments__resource_code"),
                resource_name=F("timesheet_equipments__resource_name"),
                rate=F("timesheet_equipments__rate"),
                unit=F("timesheet_equipments__unit"),
                equipment_attachments=Subquery(equipment_with_attachments[:1])
            ),
            distinct=True
        )
    )
    .values(
        "id", "timesheet_date", "clock_in_time", "clock_out_time", "timesheet_equipments"
    )
    .order_by("-timesheet_date")
    .distinct()
)

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