如何为Django查询添加全局排名注解并在过滤后保留排名？

现有如下Django模型：

class LeaderboardScore(models.Model):
    score = models.FloatField()
    player_account = models.ForeignKey(PlayerAccount, on_delete=models.CASCADE)
    timestamp = models.DateTimeField()

该模型用于记录游戏最高分，排名规则为按score降序、timestamp升序排列——分数相同时，更早的分数排名更高。

查询所有符合条件的分数时，通过Window函数可以正确添加全局排名：

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

lq = LeaderboardScore.objects.filter(leaderboard__id=1).annotate(
    rank=Window(
        expression=DenseRank(),
        order_by=[F('score').desc(), F('timestamp').asc()]
    )
)

但如果对上述查询结果按player_account过滤（例如执行lq.filter(player_account__id=123)[0].rank），返回的rank值会被重置为1，无法保留该记录的真实全局排名。

问题本质是：Window函数基于当前查询的数据集计算排名。如果先过滤再标注排名，排名会在过滤后的子集里重新计算；我们需要先在全局符合条件的数据集上计算排名，再过滤目标记录。

方法1：子查询预计算全局排名 #

通过子查询先完成全局排名的计算，外层查询再过滤目标玩家的记录：

from django.db.models import F, Window, Subquery, OuterRef
from django.db.models.functions import DenseRank

# 子查询：为所有符合条件的分数计算全局排名
rank_subquery = LeaderboardScore.objects.filter(
    leaderboard__id=1
).annotate(
    global_rank=Window(
        expression=DenseRank(),
        order_by=[F('score').desc(), F('timestamp').asc()]
    )
).filter(id=OuterRef('id')).values('global_rank')

# 外层查询：获取目标玩家的记录并携带全局排名
target_scores = LeaderboardScore.objects.filter(
    leaderboard__id=1,
    player_account__id=123
).annotate(
    rank=Subquery(rank_subquery)
)

# 输出全局排名
print(target_scores[0].rank)

方法2：内存中过滤（小数据集场景） #

如果排行榜数据量不大，可以先加载所有带全局排名的记录，再在Python内存中过滤目标玩家：

# 先获取所有带全局排名的记录
all_ranked_scores = LeaderboardScore.objects.filter(leaderboard__id=1).annotate(
    rank=Window(
        expression=DenseRank(),
        order_by=[F('score').desc(), F('timestamp').asc()]
    )
)

# 在内存中筛选目标玩家的记录
target_score = next(score for score in all_ranked_scores if score.player_account_id == 123)
print(target_score.rank)

⚠️ 注意：该方法仅适合数据量较小的场景，避免因加载大量数据导致性能瓶颈。

方法3：预计算存储排名（高并发场景） #

如果排行榜查询频率高，可将排名计算的开销转移到数据写入阶段，预计算并存储全局排名：

1. 给LeaderboardScore模型新增rank字段：

class LeaderboardScore(models.Model):
    score = models.FloatField()
    player_account = models.ForeignKey(PlayerAccount, on_delete=models.CASCADE)
    timestamp = models.DateTimeField()
    rank = models.IntegerField(null=True)  # 存储全局排名

2. 在分数新增/更新时，批量重新计算并更新排名（可通过Django信号或异步任务实现）：

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

def update_leaderboard_ranks(leaderboard_id):
    # 计算所有分数的最新全局排名
    ranked_scores = LeaderboardScore.objects.filter(leaderboard__id=leaderboard_id).annotate(
        new_rank=Window(
            expression=DenseRank(),
            order_by=[F('score').desc(), F('timestamp').asc()]
        )
    )
    # 批量更新排名字段
    for score in ranked_scores:
        score.rank = score.new_rank
        score.save()

这种方法查询时直接读取预存的rank字段，性能最优。

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