SQLAlchemy：CTE/group_by结果映射到ORM实体的问题

问题描述 #

在动态组装的复杂查询中，当特定标志开启时，需执行group_by()、sum()及group_concat()操作，但分组后无法将结果映射到ORM实体。查询本身可生成正确结果，仅映射环节失败，且需在展开步骤与连接步骤间插入分组逻辑。

场景背景 #

查询用于处理物料清单（BOM）树形结构数据，需支持4个动态步骤：

• 展开步骤：通过递归CTE获取对象直接组成项或树的叶子节点；
• 合并步骤：对相同组件分组（如合并重复电阻），此环节当前受阻；
• 连接步骤：将结果与制造商、供应商等多表关联；
• 筛选/排序步骤：对各类字段（含聚合结果）应用筛选和排序。

尝试方案 #

曾用.label()使查询结果与原表列名一致，再通过aliased(OrmEntity, subquery)或select_entity_from()映射，但ORM实体值错误，未正确获取聚合结果；也尝试为CTE新建ORM实体映射，但未找到正确方法。

相关代码 #

表结构 #

# 表结构
bom = Table('bom_details', Base.metadata,
            Column('id', Integer(), primary_key=True, autoincrement=True),
            Column('bd_bh_mag_id', Integer()),
            Column('bd_bh_revisione', String()),
            Column('bd_mag_id', Integer()),
            Column('bd_mag_id_revisione', String()),
            Column('bd_topografia', String()),
            Column('bd_flag_non_montare', Integer()),
            Column('bd_progressivo', Integer()),
            Column('bd_descrizione', String()),
            Column('bd_note', String()),
            Column('bd_qta', String()),
            ForeignKeyConstraint(['bd_bh_mag_id', 'bd_bh_revisione', 'bd_mag_id', 'bd_mag_id_revisione'],
                                 ['anamag.mag_id', 'anamag.mag_code_revision', 'anamag.mag_id', 'anamag.mag_code_revision'])
)

ORM映射 #

# ORM映射
mapper_registry.map_imperatively(
    BomItem,
    bom,
    properties={
        "descrizione" : column_property(bom.c.bd_descrizione),
        "nm": column_property(bom.c.bd_flag_non_montare),
        "index": column_property(bom.c.bd_progressivo),
        "refdes": column_property(bom.c.bd_topografia),
        "note": column_property(bom.c.bd_note),
        "qty": column_property(bom.c.bd_qta),
        "parent_anamag": relationship("AnamagItem", lazy='joined',
                               primaryjoin=(bom.c.bd_bh_mag_id == AnamagItem.iit_code) &
                               (bom.c.bd_bh_revisione == AnamagItem.revision)),
        "anamag": relationship("AnamagItem", lazy='joined',
                               primaryjoin=(bom.c.bd_mag_id == AnamagItem.iit_code) &
                               (bom.c.bd_mag_id_revisione == AnamagItem.revision)),
    },
)

合并步骤实现 #

# 合并步骤实现
def _query_imploded(self, q):
    q = q.subquery()

    subq = self.db.session.query(
        literal('0').label("bd_bh_mag_id"),
        literal('0').label("bd_bh_revisione"),
        q.c.bd_mag_id,
        q.c.bd_mag_id_revisione,
        func.group_concat(q.c.bd_topografia, separator=', ').label("bd_topografia"),
        func.group_concat(q.c.bd_flag_non_montare, separator=', ').label("bd_flag_non_montare"),
        func.group_concat(q.c.bd_progressivo, separator=', ').label("bd_progressivo"),
        literal('').label("bd_descrizione"),
        func.group_concat(q.c.bd_note, separator=', ').label("bd_note"),
        func.sum(q.c.bd_qta).label("bd_qta")
    ).group_by(q.c.bd_mag_id, q.c.bd_mag_id_revisione).cte("implode_cte")

    #res = self.db.session.query(aliased(self.db.BomItem, subq))
    #res = self.db.session.query(self.db.BomItem).select_entity_from(subq)

其中q为展开步骤生成的查询或递归CTE，最终结果需传入连接步骤进行关联、筛选和排序。

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