SQLAlchemy Core能否返回嵌套对象列表？替代func.group_concat()实现聚合

1. SQLAlchemy Core中是否有返回对象列表的类似group_concat的功能？ #

SQLAlchemy Core本身没有内置直接返回对象列表的聚合函数，但可以借助数据库原生的数组/JSON聚合函数实现类似效果，不同数据库的支持情况不同：

• PostgreSQL：array_agg()
• MySQL 8.0+：JSON_ARRAYAGG()
• SQLite 3.33+：JSON_GROUP_ARRAY()

这些函数会将聚合结果以数组或JSON格式返回，通过SQLAlchemy Core调用后，可在应用层转换成列表对象。

2. 如何在列中返回嵌套对象的数组/列表（多对多关联场景） #

以你提到的movies、actors、actors_in_movies三张表为例，以下是不同方式的实现：

纯SQL实现 #

PostgreSQL

SELECT 
    m.title,
    array_agg(a.name) AS actors
FROM movies m
JOIN actors_in_movies aim ON m.id = aim.movie_id
JOIN actors a ON a.id = aim.actor_id
GROUP BY m.id, m.title;

MySQL 8.0+

SELECT 
    m.title,
    JSON_ARRAYAGG(a.name) AS actors
FROM movies m
JOIN actors_in_movies aim ON m.id = aim.movie_id
JOIN actors a ON a.id = aim.actor_id
GROUP BY m.id, m.title;

SQLite 3.33+

SELECT 
    m.title,
    JSON_GROUP_ARRAY(a.name) AS actors
FROM movies m
JOIN actors_in_movies aim ON m.id = aim.movie_id
JOIN actors a ON a.id = aim.actor_id
GROUP BY m.id, m.title;

SQLAlchemy Core实现 #

PostgreSQL示例

from sqlalchemy import select, func
from sqlalchemy.engine import create_engine

# 假设已定义movies、actors、actors_in_movies表对象
engine = create_engine("postgresql://user:pass@localhost/db")

stmt = select(
    movies.c.title,
    func.array_agg(actors.c.name).label("actors")
).join(actors_in_movies, movies.c.id == actors_in_movies.c.movie_id
).join(actors, actors.c.id == actors_in_movies.c.actor_id
).group_by(movies.c.id, movies.c.title)

with engine.connect() as conn:
    results = conn.execute(stmt).fetchall()
    # 结果示例：
    # ("Bill & Ted's Excellent Adventure", ["Keanu Reeves", "Alex Winter", "George Carlin"])

MySQL示例

import json
from sqlalchemy import select, func

stmt = select(
    movies.c.title,
    func.json_arrayagg(actors.c.name).label("actors")
).join(actors_in_movies, movies.c.id == actors_in_movies.c.movie_id
).join(actors, actors.c.id == actors_in_movies.c.actor_id
).group_by(movies.c.id, movies.c.title)

with engine.connect() as conn:
    raw_results = conn.execute(stmt).fetchall()
    # 将JSON字符串解析为列表
    results = [(row.title, json.loads(row.actors)) for row in raw_results]

关于查询方式与处理层的选择 #

• 优先数据库层面聚合：如果你的数据库支持数组/JSON聚合函数，用单查询+数据库聚合是最优方案，性能优于应用层处理。
• 应用层处理（兼容旧数据库）：如果数据库不支持相关聚合函数，可以先查询所有电影与演员的关联数据，再用代码分组（比如Python的itertools.groupby）：

  from itertools import groupby
  
  stmt = select(
      movies.c.title,
      actors.c.name
  ).join(actors_in_movies).join(actors).order_by(movies.c.id)
  
  with engine.connect() as conn:
      rows = conn.execute(stmt).fetchall()
      # 按电影标题分组
      results = []
      for title, group in groupby(rows, key=lambda x: x.title):
          actors_list = [row.name for row in group]
          results.append((title, actors_list))
  

• 无需多查询：不管用哪种方式，都可以通过单查询完成需求，N+1查询（先查所有电影，再逐个查演员）是下选，仅在极端场景考虑。

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