如何用Django 3.2.9模型按forecast_for小时取最新forecasted_at记录

问题核心 #

需要按forecast_for的小时分组，每组取forecasted_at最新的完整InventoryForecast记录，直接在分组查询中添加count、id等字段会破坏分组逻辑，导致返回所有记录。

解决方案一：子查询关联法 #

先通过子查询按小时分组得到每组最新的forecasted_at，再将结果与主表关联，筛选出对应完整记录：

from django.db.models import Max, F
from django.db.models.functions import TruncHour

# 子查询：按forecast_for小时分组，获取每组最新的forecasted_at
latest_forecast_subquery = InventoryForecast.objects.annotate(
    forecast_for_hour=TruncHour('forecast_for')
).values('forecast_for_hour').annotate(
    latest_forecasted_at=Max('forecasted_at')
).values('forecast_for_hour', 'latest_forecasted_at')

# 关联主表，筛选出符合条件的完整记录
result = InventoryForecast.objects.annotate(
    forecast_for_hour=TruncHour('forecast_for')
).filter(
    (F('forecast_for_hour'), F('forecasted_at'))__in=latest_forecast_subquery
)

解决方案二：Window窗口函数法（推荐，PostgreSQL原生支持） #

利用Window函数给每组（按小时分组）的记录按forecasted_at倒序排名，然后筛选排名为1的记录：

from django.db.models import Window, F
from django.db.models.functions import TruncHour, RowNumber

result = InventoryForecast.objects.annotate(
    forecast_for_hour=TruncHour('forecast_for'),
    # 按小时分组，组内按forecasted_at倒序排，生成行号
    row_num=Window(
        partition_by=TruncHour('forecast_for'),
        order_by=F('forecasted_at').desc()
    )
).filter(row_num=1)

说明 #

• 方案二直接通过窗口函数在数据库层面完成分组排序和筛选，性能更优，尤其数据量较大时
• 两种方案都能返回每组最新的完整InventoryForecast对象，包含id、count等所有字段

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