这在 Django ORM 中真的不可能吗？

Posted 2023-04-12

【中文标题】这在 Django ORM 中真的不可能吗？

【英文标题】：is that really not possible in the Django ORM?

【发布时间】：2018-04-28 06:03:42

【问题描述】：

在我使用 Django 的 10 年中，我发现了一些错误，但我从未不得不退回到原始 SQL 查询。除了在这里寻求帮助之外，我几乎一整天都在摆弄 Django：

我有一个管理器应用程序，其中包含轮班和工作这些轮班的人员 (M2M)。每天还有直接存储在当天的主管（M2M）。对于用户统计数据，我想看看谁有多少班次。出于速度原因，我想只用一个查询来处理这个问题。这是我最终得到的查询：

assigned_to = self.shifts.filter(day_id=OuterRef('id')).order_by()
days = Day.data.filter(date__range=(start, end)).distinct().annotate(
    assigned_shifts=Count(
        Subquery(assigned_to.values('id'))
    )).annotate(
    assigned_pl=Max(Case(
        When(production_managers=self, then=1),
        default=0, output_field=models.IntegerField(),
        distinct=True
    ))).annotate(
    assigned_tl=Max(Case(
        When(day_managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).annotate(
    assigned_al=Max(Case(
        When(managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).annotate(
    assigned_cash=Max(Case(
        When(cash_managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).order_by()

这给了我一个 SQL 错误“作为表达式的子查询返回了不止一行”，tho。如果没有子查询，我会得到错误的值——可能是因为分组没有按预期进行。原来在文档中有一个警告和一个非常旧的票 (https://code.djangoproject.com/ticket/10060) 警告关于这个确切的场景，但不知何故我没有让子查询像它应该的那样工作。

最终的和 lil 简化的 SQL 如下所示：

days = Day.data.raw("""
    SELECT distinct
      "events_day"."id","events_day"."date",
      (select count(*) 
       from events_shift 
       left outer join events_shift_employees 
       on events_shift_employees.shift_id = events_shift.id 
       where events_shift_employees.user_id = user_id and 
       events_shift.day_id = events_day.id and 
       events_shift.deleted is null) AS "assigned_shifts",
      Max(CASE WHEN"events_day_production_managers"."user_id"=user_id THEN 1 ELSE 0 END)AS"assigned_pl",
      Max(CASE WHEN"events_day_day_managers"."user_id"=user_id THEN 1 ELSE 0 END)AS"assigned_tl",
      Max(CASE WHEN"events_day_managers"."user_id"=user_id THEN 1 ELSE 0 END)AS"assigned_al",
      max(CASE WHEN"events_day_cash_managers"."user_id"=user_id THEN 1 ELSE 0 END)AS"assigned_cash"
    FROM"events_day"
    LEFT JOIN"events_shift"ON("events_day"."id"="events_shift"."day_id")
    LEFT JOIN"events_shift_employees"ON("events_shift"."id"="events_shift_employees"."shift_id")
    LEFT JOIN"events_day_production_managers"ON("events_day"."id"="events_day_production_managers"."day_id")
    LEFT JOIN"events_day_day_managers"ON("events_day"."id"="events_day_day_managers"."day_id")
    LEFT JOIN"events_day_managers"ON("events_day"."id"="events_day_managers"."day_id")
    LEFT JOIN"events_day_cash_managers"ON("events_day"."id"="events_day_cash_managers"."day_id")
    WHERE("events_day"."deleted"IS NULL AND"events_day"."date" BETWEEN 'start' AND 'end')
    GROUP BY "events_day"."id"
    ORDER BY"events_day"."date"DESC""".format(**
        'user_id': self.id,
        'start': start,
        'end': end
    )

我可能需要一个小手来实现第 4 行中的 SELECT COUNT(*) 而不是 Count(Subquery()) Django 允许我添加到查询中，但会产生 SQL 错误。

我正在使用 Python 3.6.5、Django 2.0.3 和 Postgres 10

这个问题是核心 SQL ORM 问题，所以对于任何研究这个问题的人来说都是 THX。

【参考方案1】：

我认为这可以在没有原始 SQL 的情况下实现，只需对您的子查询稍作改动。

assigned_to = self.shifts.filter(day_id=OuterRef('id')).order_by().values('day_id')
count_assigned = assigned_to.annotate(count=Count('*')).values('count')
days = Day.data.filter(date__range=(start, end)).distinct().annotate(
    assigned_shifts=Subquery(count_assigned)
    ).annotate(
    assigned_pl=Max(Case(
        When(production_managers=self, then=1),
        default=0, output_field=models.IntegerField(),
        distinct=True
    ))).annotate(
    assigned_tl=Max(Case(
        When(day_managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).annotate(
    assigned_al=Max(Case(
        When(managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).annotate(
    assigned_cash=Max(Case(
        When(cash_managers=self, then=1),
        default=0, output_field=models.IntegerField(), distinct=True
    ))).order_by()

【讨论】：

oooohh，我没有收到这个答案的邮件，所以我今天才看到 :( 我试过你的版本，但这给了我一个需要输出字段的错误。将它添加到新的子查询根本没有输出，因此 None 被分配给assigned_shifts。当我有更多时间时，我可能会进一步调查，谢谢你的提示！