如何将含UNION的SQL查询转换为Peewee查询？

原生SQL查询 #

_, storage, spaceLeft = db.execute('''
    SELECT session, storage, storage - count(questID) FROM Hypos
    INNER JOIN Quests ON Hypos.hypoID = Quests.hypoID
    WHERE Hypos.hypoID = ?1 AND session = ?2
    GROUP BY session
    UNION SELECT NULL, max(storage), storage FROM Hypos
    WHERE Hypos.hypoID = ?1
    ORDER BY session DESC LIMIT 1
    ''', (body.hypo_id, session)
).fetchone()

尝试的Peewee代码（存在错误） #

_, storage, spaceLeft = (db_models.Hypos
        .select(db_models.Quests.session, db_models.Hypos.storage, db_models.Hypos.storage-fn.COUNT(db_models.Quests.questID))
        .join(db_models.Quests, on=(db_models.Hypos.hypoID == db_models.Quests.hypoID), attr='quests')
        .where(db_models.Hypos.hypoID==body.hypo_id, db_models.Quests.session==session)
        .group_by(db_models.Quests.session) 
    ) | (db_models.Hypos.select(None, fn.MAX(db_models.Hypos.storage), db_models.Hypos.storage)
        .where(db_models.Hypos.hypoID==body.hypo_id)
        .order_by(db_models.Quests.session.desc()).limit(1)
        )

遇到的问题 #

• 添加order_by时提示：1st ORDER BY term does not match any column in the result set
• 去掉order_by后提示：not enough values to unpack (expected 3, got 2)

错误原因与解决方法 #

错误1：排序字段不存在 #

UNION的第二个子查询中没有Quests.session字段，且排序操作应该作用于整个联合查询结果，而非单个子查询。此外，两个子查询的列名需要统一，否则排序时无法识别字段。

错误2：列数不匹配 #

第二个子查询中直接写None会被Peewee忽略，导致返回列数不足3，无法满足解包需求。需要用Value(None)显式生成NULL值，同时保证两个子查询返回的列数和别名一致。

正确的Peewee实现 #

from peewee import fn, Value

# 构造第一个子查询（对应INNER JOIN部分）
subquery1 = (db_models.Hypos
             .select(
                 db_models.Quests.session,
                 db_models.Hypos.storage,
                 (db_models.Hypos.storage - fn.COUNT(db_models.Quests.questID)).alias('space_left')
             )
             .join(db_models.Quests, on=(db_models.Hypos.hypoID == db_models.Quests.hypoID))
             .where(
                 db_models.Hypos.hypoID == body.hypo_id,
                 db_models.Quests.session == session
             )
             .group_by(db_models.Quests.session, db_models.Hypos.storage)  # 必须包含SELECT中的非聚合字段
            )

# 构造第二个子查询（对应UNION部分）
subquery2 = (db_models.Hypos
             .select(
                 Value(None).alias('session'),  # 显式生成NULL并指定别名
                 fn.MAX(db_models.Hypos.storage).alias('storage'),
                 db_models.Hypos.storage.alias('space_left')  # 与第一个子查询列别名统一
             )
             .where(db_models.Hypos.hypoID == body.hypo_id)
            )

# 联合查询、排序并获取单条结果
final_result = (subquery1 | subquery2).order_by(fn.session.desc()).limit(1).first()

# 解包结果
_, storage, spaceLeft = (final_result.session, final_result.storage, final_result.space_left)

关键说明 #

1. 列别名统一：两个子查询的列都指定相同别名，确保UNION后结果结构一致，排序和取值时不会出错。
2. NULL值处理：用Value(None)替代直接写None，保证第二个子查询返回3列数据。
3. GROUP BY规则：第一个子查询的group_by必须包含所有非聚合的SELECT字段（如Hypos.storage），避免违反SQL规范。
4. 排序位置：将order_by和limit放到整个联合查询之后，作用于合并后的完整结果集。

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