如何在SQLAlchemy中聚合多列实现账户项目信息分组查询

嘿，我来帮你搞定这个聚合需求！既然你的AWS Aurora里Account和Project是一对多关系，要把每个账户对应的project_id、project_name、document_count打包成三元组列表，这里有几个实用的实现方案，适配不同场景：

先明确表模型（基于你提到的结构） #

先确认下你的SQLAlchemy模型大概是这样的（如果有出入可以调整）：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class Account(Base):
    __tablename__ = 'account'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    projects = relationship("Project", back_populates="account")

class Project(Base):
    __tablename__ = 'project'
    id = Column(Integer, primary_key=True)
    name = Column(String, name='project_name')
    document_count = Column(Integer)
    account_id = Column(Integer, ForeignKey('account.id'))
    account = relationship("Account", back_populates="projects")

方案1：数据库层面聚合（推荐，效率最高） #

利用Aurora兼容的数据库函数（PostgreSQL/MySQL版本分别适配），直接在数据库把三个字段打包成对象再聚合成数组：

如果你用的是Aurora PostgreSQL

from sqlalchemy import func

query_result = session.query(
    Account.id.label('account_id'),
    Account.name.label('account_name'),
    # 用json_build_object打包三元组，再array_agg聚合成列表
    func.array_agg(
        func.json_build_object(
            'project_id', Project.id,
            'project_name', Project.name,
            'document_count', Project.document_count
        )
    ).label('projects')
).join(Project).group_by(Account.id, Account.name).all()

# 转成你要的字典格式
final_result = [row._asdict() for row in query_result]

如果你用的是Aurora MySQL

MySQL用JSON_OBJECT和JSON_ARRAYAGG组合：

from sqlalchemy import func

query_result = session.query(
    Account.id.label('account_id'),
    Account.name.label('account_name'),
    func.json_arrayagg(
        func.json_object(
            'project_id', Project.id,
            'project_name', Project.name,
            'document_count', Project.document_count
        )
    ).label('projects')
).join(Project).group_by(Account.id, Account.name).all()

final_result = [row._asdict() for row in query_result]

这个方案直接让数据库完成聚合，性能最优，返回的projects就是现成的三元组列表。

方案2：应用层聚合（简单易理解） #

如果不想依赖数据库特定函数，可以先拉取所有关联数据，再在Python层面聚合：

from collections import defaultdict

# 查询所有账户+关联项目的指定字段
raw_data = session.query(
    Account.id,
    Account.name,
    Project.id.label('project_id'),
    Project.name.label('project_name'),
    Project.document_count
).join(Project).all()

# 手动聚合每个账户的项目列表
account_project_map = defaultdict(list)
for acc_id, acc_name, proj_id, proj_name, doc_count in raw_data:
    account_project_map[(acc_id, acc_name)].append({
        'project_id': proj_id,
        'project_name': proj_name,
        'document_count': doc_count
    })

# 整理成期望格式
final_result = [
    {
        'account_id': acc_id,
        'account_name': acc_name,
        'projects': projects
    } for (acc_id, acc_name), projects in account_project_map.items()
]

这个方案适合数据量不大的场景，代码直观，不需要记数据库函数。

方案3：ORM复合类型聚合（优雅的类型化方式） #

如果希望返回的是强类型的复合对象，可以定义SQLAlchemy复合类型：

from sqlalchemy import CompositeType, Column, Integer, String

# 定义项目三元组的复合类型
ProjectTuple = CompositeType(
    'project_tuple',
    [
        Column('project_id', Integer),
        Column('project_name', String),
        Column('document_count', Integer)
    ]
)

# 执行聚合查询
query_result = session.query(
    Account.id.label('account_id'),
    Account.name.label('account_name'),
    func.array_agg(
        func.cast(
            func.row(Project.id, Project.name, Project.document_count),
            ProjectTuple
        )
    ).label('projects')
).join(Project).group_by(Account.id, Account.name).all()

# 转成字典格式（如果需要）
final_result = []
for row in query_result:
    projects = [{'project_id': p.project_id, 'project_name': p.project_name, 'document_count': p.document_count} for p in row.projects]
    final_result.append({
        'account_id': row.account_id,
        'account_name': row.account_name,
        'projects': projects
    })

这个方案返回的projects是类型化的对象，代码更优雅，适合大型项目的类型规范。

额外优化：保留无项目的账户 #

如果要包含没有任何项目的账户，把join换成outerjoin，并用coalesce把空聚合转成空数组：

# PostgreSQL示例
func.coalesce(func.array_agg(...), '[]'::json[]).label('projects')
# MySQL示例
func.coalesce(func.json_arrayagg(...), '[]').label('projects')

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