SqlAlchemy的array_agg函数是否支持多列或整行聚合？

在PostgreSQL中可以通过原生SQL实现聚合整行数据到数组：

SELECT 
     array_agg(users.*) 
FROM users 
WHERE 
     l_name LIKE 'Br%' 
GROUP BY f_name;

但用SQLAlchemy实现时，会抛出报错：

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'InstrumentedAttribute'

尝试仅聚合单列（如下）可以正常运行，但只能返回单列值，无法满足整行数据的需求：

query: Query[User] = session.query(array_agg(self.user.f_name))

显式指定多列组成元组的方式也会触发相同报错：

query: Query[User] = session.query(array_agg((self.user.f_name, self.user.l_name))))

方法1：直接传递模型实体到array_agg（SQLAlchemy 1.4+） #

SQLAlchemy 1.4及以上版本支持直接将模型实体传入array_agg，会自动转换为SQL中的users.*：

from sqlalchemy import func

# 生成聚合整行数据的查询
query = session.query(func.array_agg(self.user))\
               .filter(self.user.l_name.like('Br%'))\
               .group_by(self.user.f_name)

# 执行查询并获取结果
results = query.all()
# 每个结果中的数组元素会映射为User对象
for row in results:
    aggregated_users = row[0]
    for user in aggregated_users:
        print(user.f_name, user.l_name)

方法2：使用func.row()包裹所有列（兼容旧版本） #

如果使用的是SQLAlchemy旧版本，或者需要更明确地控制行结构，可以用func.row()包裹模型的所有列，生成数据库层面的行构造器：

from sqlalchemy import func

# 获取模型的所有列
user_columns = self.user.__table__.columns

# 聚合整行数据
query = session.query(func.array_agg(func.row(*user_columns)))\
               .filter(self.user.l_name.like('Br%'))\
               .group_by(self.user.f_name)

# 执行查询后，可将行数据转换为User对象
results = query.all()
for row in results:
    user_rows = row[0]
    user_objects = [User(**dict(zip(self.user.__table__.columns.keys(), ur))) for ur in user_rows]

方法3：结合cast实现类型映射 #

如果需要确保聚合结果能正确映射为模型对象，可以用cast将行数据转换为PostgreSQL的RECORD类型：

from sqlalchemy import func, cast, ARRAY
from sqlalchemy.dialects.postgresql import RECORD

query = session.query(
    func.array_agg(cast(self.user, RECORD)).label('aggregated_users')
).filter(self.user.l_name.like('Br%')).group_by(self.user.f_name)

results = query.all()
for row in results:
    user_objects = [User(**dict(zip(User.__table__.columns.keys(), u))) for u in row.aggregated_users]

报错原因说明 #

之前尝试的(self.user.f_name, self.user.l_name)元组写法，会被SQLAlchemy识别为Python本地元组，而非PostgreSQL的行构造器，导致psycopg2无法适配InstrumentedAttribute类型。必须使用SQLAlchemy提供的func.row()或直接传递模型实体，才能生成正确的SQL行表达式。

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