使用SQLAlchemy的group_by()按id分组获取条目最高优先级

实现方法 #

你可以通过CASE表达式配合聚合函数实现这个分组优先级判定，逻辑直接在数据库层执行，不需要拉取全量数据到内存处理，兼容MySQL、SQLite、PostgreSQL等所有主流数据库。

ORM 版本示例

假设你的映射模型类名为Record，分组字段是id，存储优先级值的字段为priority：

from sqlalchemy import func, case

# 为不同优先级分配权重，数值越高优先级越高
priority_weight = case(
    (Record.priority == "high", 3),
    (Record.priority == "medium", 2),
    (Record.priority == "low", 1),
    else_=0
)

# 分组后取最大权重，映射回对应优先级文本
priority_res = case(
    (func.max(priority_weight) == 3, "high"),
    (func.max(priority_weight) == 2, "medium"),
    (func.max(priority_weight) == 1, "low"),
    else_="no priority found"
).label("priority")

# 执行查询
results = db_session.query(Record.id, priority_res).group_by(Record.id).all()

Core 版本示例

如果使用Core语法，表对象为record_table：

from sqlalchemy import func, case, select

priority_weight = case(
    (record_table.c.priority == "high", 3),
    (record_table.c.priority == "medium", 2),
    (record_table.c.priority == "low", 1),
    else_=0
)

priority_res = case(
    (func.max(priority_weight) == 3, "high"),
    (func.max(priority_weight) == 2, "medium"),
    (func.max(priority_weight) == 1, "low"),
    else_="no priority found"
).label("priority")

stmt = select(record_table.c.id, priority_res).group_by(record_table.c.id)
results = db_session.execute(stmt).all()

逻辑说明 #

权重映射规则完全匹配需求：分组内只要存在high，最大权重就为3，直接返回high；没有high时如果存在medium，最大权重为2，返回medium；前两者都没有但存在low时返回low；其余情况返回默认值。
查询返回的结果为(id, priority)格式的元组列表，和期望输出结构完全一致。

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