如何使用带有 string_agg 的子查询插入带有 SQLAlchemy ORM 的行？答案

【问题标题】：How to insert row with SQLAlchemy ORM using a subquery with string_agg?如何使用带有 string_agg 的子查询插入带有 SQLAlchemy ORM 的行？
【发布时间】：2021-01-18 15:30:41
【问题描述】：

我有一个向表中插入一行的 SQL 查询。其中一列分配了一个从使用string_agg 聚合多个值的子查询返回的值。如何使用 SQLAlchemy ORM 编写？

postgres 查询

INSERT INTO blobs (_id, data) VALUES (nextval('blobs_id_seq'),
  (SELECT string_agg(blobs.data, NULL ORDER BY blobs._id DESC) 
    FROM blobs where blobs._id IN (1,2) ) );

表 ORM

    class Blob(Base):
      __tablename__ = 'blobs'
      _id = Column(Integer, Sequence('blobs_id_seq'), primary_key=True, unique=True)
     data = Column(LargeBinary)

    Base.metadata.create_all(engine)

    db = Session(engine)
    db.add(Blob(data=b'aa'))
    db.add(Blob(data=b'bb'))
    db.add(Blob(data=b'cc'))
    db.commit()

无效的 ORM 插入

    # Should yield a "Blob" with data = b"bbaa"
    db.add(Blob(data=db.query(func.string_agg(Blob.data, None))
      .filter( Blob._id.in_([1, 2]) )
      .order_by( Blob._id.desc() ) ))

【问题讨论】：

标签： python postgresql orm sqlalchemy

【解决方案1】：

事实证明，有一个特定于方言的帮助器 aggregate_order_by 可以呈现输出预期结果的查询。

db.add(Blob(data=db.query(func.string_agg(Blob.data,
    aggregate_order_by(None, Blob._id.desc())))
  .filter( Blob._id.in_([1, 2]) )))

【讨论】：