如何在SQLAlchemy中用array_agg()关联查询获取Product表数据数组？

我希望使用array_agg函数获取Product表的数据数组，该功能在PostgreSQL中可正常运行，但在SQLAlchemy中似乎仅支持整数、字符串等基础数据类型，请问如何在SQLAlchemy中实现该功能？

PostgreSQL示例代码 #

SELECT DISTINCT
    seller.title, array_agg(product), COUNT(product.id)
FROM seller_product
    INNER JOIN seller ON seller.ozon_id = seller_product.id_seller 
    INNER JOIN product ON product.ozon_id = seller_product.id_product 
WHERE start_id = 36
GROUP BY seller.title ORDER BY COUNT(product.id) DESC

我尝试的Python代码 #

slct_stmt_now = select(Seller, func.array_agg(Product.__table__), func.count(Product.id)).distinct().select_from(seller_product)
slct_stmt_now = slct_stmt_now.join(Seller, Seller.ozon_id == seller_product.columns["id_seller"]).join(Product, Product.ozon_id == seller_product.columns["id_product"]).where(seller_product.columns["start_id"] == LAST_START_ID).group_by(Seller)
now_data_txt = session.execute(slct_stmt_now.order_by(func.count(Product.id).desc())).all()

在SQLAlchemy中实现PostgreSQL的array_agg聚合整个行对象，按以下方式调整代码即可：

• 替换聚合对象为实体而非表结构
  不要传入Product.__table__，直接使用Product实体，SQLAlchemy会自动将其解析为对应表的行记录，聚合部分改为func.array_agg(Product)。

• 移除冗余的distinct()
  原生SQL中的DISTINCT是多余的，GROUP BY seller.title已经会对卖家分组并合并重复行，SQLAlchemy查询中可以去掉.distinct()。

• 对齐分组字段与原生SQL
  原生SQL按seller.title分组，代码中用group_by(Seller)会按Seller所有字段分组（若表主键为单字段可能可行，但更严谨的写法是group_by(Seller.title)）。

修改后的Python代码：

slct_stmt_now = (
    select(
        Seller.title,
        func.array_agg(Product),
        func.count(Product.id)
    )
    .select_from(seller_product)
    .join(Seller, Seller.ozon_id == seller_product.columns["id_seller"])
    .join(Product, Product.ozon_id == seller_product.columns["id_product"])
    .where(seller_product.columns["start_id"] == LAST_START_ID)
    .group_by(Seller.title)
    .order_by(func.count(Product.id).desc())
)
now_data_txt = session.execute(slct_stmt_now).all()

• 可选：聚合为JSON数组
  如果需要直接得到可序列化的JSON格式数组，可以结合PostgreSQL的row_to_json函数，将聚合结果转为JSON数组：

  slct_stmt_now = select(
      Seller.title,
      func.array_agg(func.row_to_json(Product)),
      func.count(Product.id)
  )
  # 后续连接、过滤、分组逻辑同上
  

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