如何用Django ORM单查询计算产品平均配送时长（天）

首先修正模型中的笔误：models.ForiegnKey 应该改为 models.ForeignKey，否则会导致数据库迁移错误。

要实现单查询计算所有产品的平均配送时长，我们可以利用Django ORM的子查询（Subquery）、**注解（annotate）和聚合（aggregate）**功能，避免循环遍历所有记录。以下是具体实现：

步骤说明 #

1. 使用子查询分别获取每个产品的「发货时间（in_transit）」和「签收时间（delivered）」；
2. 过滤掉缺少任一状态事件的产品（确保计算的有效性）；
3. 计算单个产品的配送时长（天数）；
4. 对所有有效产品的配送时长求平均值。

代码实现 #

from django.db.models import Subquery, OuterRef, Avg, F
from django.db.models.functions import ExtractDay

# 子查询：获取每个产品的in_transit事件时间（假设每个产品仅一条该状态记录）
in_transit_subquery = ProductEvents.objects.filter(
    product=OuterRef('pk'),
    status=ProductEvents.Status.IN_TRANSIT
).values('created')[:1]

# 子查询：获取每个产品的delivered事件时间
delivered_subquery = ProductEvents.objects.filter(
    product=OuterRef('pk'),
    status=ProductEvents.Status.DELIVERED
).values('created')[:1]

# 计算平均配送时长（天）
average_delivery_days = Product.objects.annotate(
    # 为每个产品添加两个时间字段
    in_transit_date=Subquery(in_transit_subquery),
    delivered_date=Subquery(delivered_subquery)
).filter(
    # 过滤掉缺少任一事件的产品
    in_transit_date__isnull=False,
    delivered_date__isnull=False
).annotate(
    # 计算单产品配送时长（天数）
    duration_days=ExtractDay(F('delivered_date') - F('in_transit_date'))
).aggregate(
    # 求所有产品的平均时长
    avg_days=Avg('duration_days')
)

# 结果示例：{'avg_days': 2.7}
print(average_delivery_days)

注意事项 #

• 如果产品可能存在多条同状态的事件记录（比如多次发货/签收），请在子查询中添加order_by来指定取哪一条记录，例如取最新的事件：

  in_transit_subquery = ProductEvents.objects.filter(
      product=OuterRef('pk'),
      status=ProductEvents.Status.IN_TRANSIT
  ).order_by('-created').values('created')[:1]
  

• 上述代码使用ExtractDay直接在数据库层面提取天数，性能更优；如果需要更精确的时长（包含小时/分钟），可以保留DurationField类型的时间差，后续在Python中通过timedelta.total_seconds() / 86400转换为天数。

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