如何用Django ORM可靠获取各用户最新记录并筛选过期项

数据模型与示例数据 #

模型定义 #

class RdpUserPasswordHash(models.Model):
    user = models.ForeignKey(get_user_model(), on_delete=models.CASCADE, related_name='password_hash')
    when = models.DateTimeField(auto_now_add=True, db_index=True)

用户ID=3的示例记录 #

1 2023-07-29 10:55:41.193 +0200  3
2 2022-08-09 10:55:41.236 +0200  3

需求与问题 #

需要通过Django ORM实现：

• 获取每个用户对应的最新when日期的记录
• 从中筛选出已过期（when早于指定阈值）的记录

尝试过order_by('user_id', '-when').distinct('user_id')可以正确获取各用户最新记录，但添加filter(when__lt=timezone.now() - td_expire_days)后结果不符合预期——返回的是用户过期记录中的最新条目，而非用户所有记录里的最新条目是否过期。

临时方案只能拿到用户ID和最新日期，需要循环查询才能获取完整的RdpUserPasswordHash实例：

expired_users = RdpUserPasswordHash.objects.values('user_id').annotate(latest_date=Max('when')).filter(latest_date__lt=timezone.now() - td_expire_days)

最优ORM实现方案 #

使用子查询（Subquery）+ OuterRef可以直接获取符合条件的完整模型实例，无需循环查询：

代码实现 #

from django.db.models import Subquery, OuterRef, Max
from django.utils import timezone

# 设置过期天数阈值，根据实际业务调整
td_expire_days = timezone.timedelta(days=90)
cutoff_time = timezone.now() - td_expire_days

# 子查询：获取每个用户的最新when时间戳
latest_when_subquery = RdpUserPasswordHash.objects.filter(
    user_id=OuterRef('user_id')
).values('user_id').annotate(latest_when=Max('when')).values('latest_when')

# 主查询：匹配最新时间戳，同时过滤过期记录
expired_password_hashes = RdpUserPasswordHash.objects.filter(
    when=Subquery(latest_when_subquery),
    when__lt=cutoff_time
)

原理说明 #

1. 子查询先计算出每个用户的最新when值，确保我们拿到的是用户所有记录里的最新条目
2. 主查询通过when=Subquery(...)匹配到该用户的最新记录，再通过when__lt=cutoff_time筛选出已过期的条目
3. 最终返回的是完整的RdpUserPasswordHash实例集合，可直接使用，无需额外查询

为什么之前的方法失效？ #

Django QuerySet的执行顺序是先过滤（filter），再排序（order_by），最后去重（distinct）。直接在order_by+distinct前加filter，会先把未过期的记录过滤掉，剩下的过期记录里取每个用户的最新，这和需求（先取用户所有记录的最新，再判断是否过期）逻辑不符。

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