如何在SQLAlchemy中实现带筛选关联子查询并引用外部字段？

需求原生SQL #

用户需要实现的原生SQL逻辑如下：

SELECT men.judul, 
  (SELECT 
       group_concat(m_trainer.nama_karyawan ORDER BY m_trainer.nama_karyawan) AS nama_trainer
    FROM m_detail_kelas_admin
         INNER JOIN m_kelas_admin
            ON     m_kelas_admin.id_kelas = m_detail_kelas_admin.id_kelas
               AND m_kelas_admin.status_aktif = 1
         INNER JOIN m_materi_admin
            ON     m_materi_admin.id_materi = m_detail_kelas_admin.id_materi
               AND m_materi_admin.status_aktif = 1
         INNER JOIN m_trainer
            ON     FIND_IN_SET(m_trainer.nik, m_materi_admin.trainer) > 0
    WHERE men.id_kelas = m_kelas_admin.id_kelas) AS list_guru
FROM m_kelas_admin men;

用户错误尝试代码 #

用户编写的SQLAlchemy代码无法正确关联主查询字段，核心问题是子查询中重复关联了主查询表，导致无法区分主查询的kelas.id_kelas：

detail_kelas = model_kelas_admin.m_detail_kelas_admin
kelas = model_kelas_admin.m_kelas_admin
materi = model_materi_admin.m_materi_admin
trainer = model_activity.m_trainer
subquery_trainer = db.session.query(detail_kelas).\
    join(kelas, and_(kelas.id_kelas == detail_kelas.id_kelas, kelas.status_aktif == 1)).\
    join(materi, and_(materi.id_materi == detail_kelas.id_materi, materi.status_aktif == 1)).\
    join(trainer, and_(func.find_in_set(trainer.nik, materi.trainer), trainer.status_aktif == 1)).\
    filter(
      detail_kelas.status_aktif == "1",
      detail_kelas.id_kelas == kelas.id_kelas -- 此处引用的是子查询内部的kelas，不是主查询的
    ).with_entities(
      func.group_concat(trainer.nama_karyawan).label('nama_trainer')
    ).scalar_subquery()

data = db.session.query(kelas).\
    filter(
      kelas.status_aktif == "1"
    ).with_entities(
      kelas.id_kelas,
      kelas.judul,
      subquery_trainer.label('list_trainer')
    ).all()

正确实现方案 #

核心解决思路是给主查询表创建别名，让子查询明确引用主查询的字段，实现关联子查询（Correlated Subquery）：

from sqlalchemy import and_, func

# 模型引用
detail_kelas = model_kelas_admin.m_detail_kelas_admin
kelas = model_kelas_admin.m_kelas_admin
materi = model_materi_admin.m_materi_admin
trainer = model_activity.m_trainer

# 给主查询的kelas表创建别名，对应原生SQL中的men
men = kelas.alias('men')

# 定义关联子查询：明确引用主查询的men.id_kelas
subquery_trainer = (
    db.session.query(
        # 保持原生SQL的group_concat排序逻辑
        func.group_concat(trainer.nama_karyawan.order_by(trainer.nama_karyawan)).label('nama_trainer')
    )
    # 按原生SQL的关联顺序拼接表
    .join(kelas, and_(kelas.id_kelas == detail_kelas.id_kelas, kelas.status_aktif == 1))
    .join(materi, and_(materi.id_materi == detail_kelas.id_materi, materi.status_aktif == 1))
    .join(trainer, func.find_in_set(trainer.nik, materi.trainer) > 0)
    # 关键：关联子查询与主查询的kelas.id_kelas
    .filter(kelas.id_kelas == men.id_kelas)
    .scalar_subquery()
)

# 主查询使用别名men，关联子查询
data = (
    db.session.query(
        men.id_kelas,
        men.judul,
        subquery_trainer.label('list_trainer')
    )
    .filter(men.status_aktif == "1")
    .all()
)

关键说明 #

1. 表别名区分：通过kelas.alias('men')给主查询表创建独立别名，避免和子查询中关联的kelas表冲突；
2. 关联子查询逻辑：子查询的filter(kelas.id_kelas == men.id_kelas)直接关联主查询的字段，和原生SQL的WHERE men.id_kelas = m_kelas_admin.id_kelas完全等价；
3. 保持原生逻辑：group_concat保留order_by排序，关联条件完全对齐原生SQL的ON和WHERE规则；
4. 冗余清理：移除用户代码中多余的detail_kelas.status_aktif == "1"和trainer.status_aktif == 1（原生SQL未包含，若业务需要可自行添加）。

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