如何在SQLAlchemy中一次查询某列的上下邻近值？

我完全懂你的困扰——现在要分开调用两个函数才能拿到目标XP值的上下邻近成就，不仅麻烦，还在子查询、表连接这些用法上摸不清方向。其实咱们可以用SQLAlchemy把两个查询合并成一次，既减少数据库交互，也能理清子查询的用法。

方案1：用UNION合并两个子查询（通用所有数据库） #

这个方法最直接，把你原来的两个查询改成子查询，然后用UNION ALL合并结果，一次会话就能拿到两个目标：

def get_nearby_XP_goals(xp):
    session = DbSessionFactory.create_session()
    
    # 子查询1：获取下一个更高的XP目标（只取第一行）
    next_subquery = session.query(XPAchievement) \
        .filter(XPAchievement.XP_required > xp) \
        .order_by(XPAchievement.XP_required.asc()) \
        .limit(1) \
        .subquery(name="next_goal")
    
    # 子查询2：获取上一个（包含当前）的XP目标（只取第一行）
    prev_subquery = session.query(XPAchievement) \
        .filter(XPAchievement.XP_required <= xp) \
        .order_by(XPAchievement.XP_required.desc()) \
        .limit(1) \
        .subquery(name="prev_goal")
    
    # 合并两个子查询的结果
    combined_results = session.query(XPAchievement) \
        .from_statement(
            "SELECT * FROM next_goal UNION ALL SELECT * FROM prev_goal"
        ).all()
    
    # 整理结果，区分前后目标
    previous_goal = None
    next_goal = None
    for goal in combined_results:
        if goal.XP_required <= xp:
            previous_goal = goal
        else:
            next_goal = goal
    
    return {"previous": previous_goal, "next": next_goal}

方案2：用窗口函数（适合支持窗口函数的数据库，比如PostgreSQL、MySQL 8+） #

如果你的数据库支持窗口函数，这个方法更高效，不需要两次子查询，直接通过窗口函数LAG()和LEAD()获取相邻值：

from sqlalchemy import func

def get_nearby_XP_goals_with_window(xp):
    session = DbSessionFactory.create_session()
    
    # 第一步：给所有XP成就排序，用窗口函数标记每个成就的前后XP值
    ranked_achievements = session.query(
        XPAchievement,
        # 获取当前成就的上一个XP值
        func.lag(XPAchievement.XP_required).over(order_by=XPAchievement.XP_required).label("prev_xp"),
        # 获取当前成就的下一个XP值
        func.lead(XPAchievement.XP_required).over(order_by=XPAchievement.XP_required).label("next_xp")
    ).subquery()
    
    # 第二步：找到最接近目标xp的那个成就（即<=xp的最大XP值）
    target_row = session.query(ranked_achievements) \
        .filter(ranked_achievements.c.XP_required <= xp) \
        .order_by(ranked_achievements.c.XP_required.desc()) \
        .first()
    
    # 第三步：根据前后XP值获取对应的完整成就对象（如果需要的话）
    previous_goal = None
    next_goal = None
    if target_row:
        if target_row.prev_xp:
            previous_goal = session.query(XPAchievement).filter(XPAchievement.XP_required == target_row.prev_xp).first()
        if target_row.next_xp:
            next_goal = session.query(XPAchievement).filter(XPAchievement.XP_required == target_row.next_xp).first()
    
    return {"previous": previous_goal, "next": next_goal}

小提示 #

你原来的函数里有个拼写错误：xp_gaol应该是xp_goal，记得修正哦！

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