Django如何基于两列查询数据库最新记录以获取指定挑战所有用户最新积分

你的现有代码存在N+1查询问题，循环内每次查询数据库，数据量增长后接口性能会大幅下降，以下提供3种不同场景的高效实现方案：

方案1：窗口函数实现（Django 2.0+支持，全数据库兼容，推荐） #

仅需1次数据库查询，性能最优：

from django.db.models import F, Window
from django.db.models.functions import Rank

def challenge_leaderboard(request):
    challenge_id = request.GET.get('challenge_id')
    # 对每个用户的记录按创建时间倒序排名
    ranked_logs = RewardPointLog.objects.filter(
        challenge_user__challenge_id=challenge_id
    ).annotate(
        rank=Window(
            expression=Rank(),
            partition_by=[F('challenge_user__user_id')],
            order_by=F('created_datetime').desc()
        )
    ).filter(rank=1) # 仅取每个用户排名第一的最新记录
    
    # 整理为你需要的返回格式
    user_points = {log.challenge_user.user_id: log.points for log in ranked_logs}
    # 如需返回CH1, U1 - PV1格式的列表，替换为以下逻辑：
    # user_points = [f"{challenge_id}, U{log.challenge_user.user_id} - {log.points}" for log in ranked_logs]
    return Response(user_points, status=200)

方案2：distinct去重实现（仅PostgreSQL数据库支持） #

如果使用PostgreSQL作为数据库，可以用更简洁的写法：

def challenge_leaderboard(request):
    challenge_id = request.GET.get('challenge_id')
    latest_logs = RewardPointLog.objects.filter(
        challenge_user__challenge_id=challenge_id
    ).order_by(
        'challenge_user__user_id', '-created_datetime'
    ).distinct('challenge_user__user_id')
    
    user_points = {log.challenge_user.user_id: log.points for log in latest_logs}
    return Response(user_points, status=200)

注意：带字段参数的distinct是PostgreSQL专属特性，MySQL、SQLite不支持该用法。

方案3：子查询实现（兼容所有数据库，适合Django 2.0以下版本） #

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

def challenge_leaderboard(request):
    challenge_id = request.GET.get('challenge_id')
    # 子查询获取每个用户的最新记录创建时间
    latest_time_subq = RewardPointLog.objects.filter(
        challenge_user__challenge_id=challenge_id,
        challenge_user__user_id=OuterRef('challenge_user__user_id')
    ).values('challenge_user__user_id').annotate(
        latest_time=Max('created_datetime')
    ).values('latest_time')[:1]
    
    # 关联子查询过滤出所有用户的最新记录
    latest_logs = RewardPointLog.objects.filter(
        challenge_user__challenge_id=challenge_id,
        created_datetime=Subquery(latest_time_subq)
    )
    user_points = {log.challenge_user.user_id: log.points for log in latest_logs}
    return Response(user_points, status=200)

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