如何在SQLAlchemy中使用array_agg获取多字段的列表/元组结果

问题场景 #

使用PostgreSQL + SQLAlchemy时，想要通过array_agg聚合多字段组合数据，但直接编写func.array_agg((Task.id, Task.user_id))会报错，添加func.distinct后得到的是字符串形式的聚合结果：

(100, '{"(91,1)","(92,1)","(93,1)","(94,1)"}')
(200, '{"(95,1)","(96,1)","(97,1)","(98,1)","(99,1)"}')

期望得到字符串列表，或更理想的元组列表，尝试指定type_=ARRAY(TEXT)后反而得到拆分的字符数组，需要解决该问题并移除func.distinct。

解决方案1：聚合为元组列表（推荐） #

PostgreSQL支持聚合行类型，SQLAlchemy中可以用func.row()包装多个字段，配合array_agg实现元组列表的聚合，同时通过类型映射让结果直接解析为Python元组：

from sqlalchemy import func, ARRAY, TupleType, Integer
from your_module import Task, session  # 替换为你的模型和会话

# 按分组字段聚合多字段元组
query = (
    session.query(
        Task.task_group_id,
        func.array_agg(
            func.row(Task.id, Task.user_id),
            type_=ARRAY(TupleType(Integer, Integer))
        ).label("task_pairs")
    )
    .group_by(Task.task_group_id)
)

# 执行查询
results = query.all()

执行后得到的结果符合预期：

(100, [(91,1),(92,1),(93,1),(94,1)])
(200, [(95,1),(96,1),(97,1),(98,1),(99,1)])

这里无需func.distinct，直接聚合行类型即可，PostgreSQL会正确识别多字段组合。

解决方案2：聚合为字符串列表 #

如果只需要"(id,user_id)"格式的字符串列表，可以用func.concat拼接字段，再聚合为TEXT数组：

from sqlalchemy import func, ARRAY, Text
from your_module import Task, session

query = (
    session.query(
        Task.task_group_id,
        func.array_agg(
            func.concat('(', Task.id, ',', Task.user_id, ')'),
            type_=ARRAY(Text)
        ).label("task_str_pairs")
    )
    .group_by(Task.task_group_id)
)

results = query.all()

执行后得到：

(100, ["(91,1)","(92,1)","(93,1)","(94,1)"])
(200, ["(95,1)","(96,1)","(97,1)","(98,1)","(99,1)"])

错误原因分析 #

• 直接写func.array_agg((Task.id, Task.user_id))时，SQLAlchemy会将括号内的字段解析为逗号分隔的字符串，而非PostgreSQL的行类型，因此聚合结果是字符串形式。
• 指定type_=ARRAY(TEXT)但未正确包装聚合内容时，SQLAlchemy会把整个聚合后的字符串（如'{"(91,1)",...}'）当作单个TEXT值，进而拆分成字符数组。

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