如何仅用SQL和Peewee库实现用户XP排名查询？

我定义了用于用户表的Peewee类：

class User(peewee.Model):
    """Class for Telegram user"""

    user_id = peewee.BigIntegerField(unique=True)
    xp = peewee.IntegerField(default=0)

同时编写了一个查询用户XP排名的函数：

async def get_user_position(self, user_id):
    """Get user position"""
        
    users = await orm.execute(
         (
             User.select(User.user_id)
            .group_by(User.user_id)
            .order_by(User.xp.desc()))
        .dicts())
        
    users = list(users)
        
    position = next((i+1 for i, player in enumerate(users) if player['user_id'] == user_id), None)
        
    if position == None:
        position = len(users) + 1
        
    return position

请问如何仅使用SQL语句（不使用next函数），通过Peewee库实现该功能？

可以利用SQL的窗口函数在数据库层面直接计算用户排名，无需将全量用户数据加载到内存处理。以下是改造后的异步函数：

from peewee import fn

async def get_user_position(self, user_id):
    """Get user position using SQL window function"""
    # 定义窗口：按XP降序生成连续排名
    rank_window = fn.RowNumber().over(order_by=[User.xp.desc()]).alias('position')
    
    # 组合查询：优先返回目标用户的排名，不存在则返回总用户数+1
    query = (
        User.select(rank_window)
        .where(User.user_id == user_id)
        .union_all(
            User.select(fn.COUNT(User.user_id) + 1)
            .having(fn.COUNT(User.user_id) >= 0)  # 确保此分支始终可执行
        )
        .limit(1)
    )
    
    result = await orm.execute(query)
    return next(iter(result))[0]

关键说明： #

• 窗口函数RowNumber()：数据库会直接为每个用户按XP降序分配唯一排名，避免了原方案中遍历全量用户的开销，性能更优。
• 用户不存在的处理：通过UNION_ALL拼接两个查询，当第一个查询（找目标用户排名）无结果时，第二个查询会返回当前总用户数加1，符合原逻辑。
• 简化执行：用limit(1)确保只返回一条结果，直接提取数值即可。

如果你的数据库不支持窗口函数（如旧版MySQL），可以用子查询实现：

async def get_user_position(self, user_id):
    """Get user position using subquery (for older databases)"""
    # 统计XP高于目标用户的人数，加1即为排名
    rank_subquery = User.select(User.xp).where(User.user_id == user_id)
    rank_query = User.select(fn.COUNT(User.user_id) + 1).where(User.xp > rank_subquery)
    
    # 检查用户是否存在
    user_exists = await orm.execute(User.select().where(User.user_id == user_id).exists())
    if not user_exists:
        total_count = await orm.execute(User.select(fn.COUNT(User.user_id)))
        return total_count[0][0] + 1
    
    result = await orm.execute(rank_query)
    return result[0][0]

子查询方案说明： #

• 此方案通过统计比目标用户XP高的用户数量来计算排名，注意：如果存在多个用户XP相同，他们会获得相同的排名（和窗口函数ROW_NUMBER()的连续排名逻辑不同，若需要相同XP同排名，可改用窗口函数RANK()）。

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