PostgreSQL分组查询转SQLAlchemy Asyncio及双列结果获取问题

错误原因分析 #

1. Order By逻辑错误：原代码直接按BaseModel.created_at排序，但分组后该列未在GROUP BY中，也未使用聚合函数，违反SQL分组规则。
2. 结果提取方式错误：scalars().all()仅提取查询结果的第一列，导致无法获取max(created_at)的值。
3. 子句顺序错误：SQLAlchemy中filter需在group_by之前执行，group_by要在order_by之前，与SQL语法逻辑一致。

修正后的代码实现 #

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import func

# 构建模糊查询条件
query_pattern = "%{}%".format(query)

# 定义聚合表达式并添加别名
max_created_at = func.max(BaseModel.created_at).label("max_1")

# 执行异步查询
result_set = await session.execute(
    select(BaseModel.xml_uri, max_created_at)
    .filter(BaseModel.xml_uri.like(query_pattern))
    .group_by(BaseModel.xml_uri)
    .order_by(max_created_at.desc())
)

# 获取完整结果（两种方式可选）
# 方式1：元组列表，每个元组包含(xml_uri, max_1)
records = result_set.all()

# 方式2：字典映射列表，可通过键名访问值
# records = result_set.mappings().all()

关键修正点说明 #

• 聚合表达式别名：将func.max(BaseModel.created_at)标记为max_1，既对应原SQL的别名，也方便在order_by中直接引用，避免重复编写聚合函数。
• 子句顺序调整：遵循select → filter → group_by → order_by的顺序，符合SQL执行逻辑。
• 结果提取优化：使用all()获取完整行数据（元组形式），或mappings().all()获取字典映射，确保能同时拿到xml_uri和max(created_at)两列的值。

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