SQLAlchemy中如何用join式查询实现关联字段预加载并添加约束？

现有如下SQLAlchemy模型：

class A(Base):
    __tablename__ = "as"

    id = mapped_column(Integer, primary_key=True)
    b_id = mapped_column(ForeignKey("bs.id"))
    b: Mapped[B] = relationship()


class B(Base):
    __tablename__ = "bs"

    id = mapped_column(Integer, primary_key=True)
    c_id = mapped_column(ForeignKey("cs.id"))
    c: Mapped[C] = relationship()
    x = mapped_column(Integer)


class C(Base):
    __tablename__ = "cs"

    id = mapped_column(Integer, primary_key=True)
    y = mapped_column(Integer)

需要查询满足关联字段.b.x和.b.c.y约束条件的A对象，且要求查询结果中的A对象关联字段已预加载（非延迟加载）。尝试了四种方法但都存在问题：

• 使用joinedload直接加约束：结果不正确

  select(A)
    .options(joinedload(A.b).joinedload(B.c))
    .where(B.x == 0, C.y == 0)
  

  生成的SQL：

  SELECT `as`.id, `as`.b_id, cs_1.id AS id_1, cs_1.y, bs_1.id AS id_2, bs_1.c_id, bs_1.x
  FROM `as` LEFT OUTER JOIN bs AS bs_1 ON bs_1.id = `as`.b_id LEFT OUTER JOIN cs AS cs_1 ON cs_1.id = bs_1.c_id, bs, cs
  WHERE bs.x = ? AND cs.y = ?
  

  约束未作用于关联的连接表，导致结果错误。

• 使用.has()方法：查询低效

  select(A)
    .options(joinedload(A.b).joinedload(B.c))
    .where(
      A.b.has(
        and_(B.x == 0, B.c.has(C.y == 0))
      )
    )
  

  生成的SQL：

  SELECT `as`.id, `as`.b_id, cs_1.id AS id_1, cs_1.y, bs_1.id AS id_2, bs_1.c_id, bs_1.x
  FROM `as` LEFT OUTER JOIN bs AS bs_1 ON bs_1.id = `as`.b_id LEFT OUTER JOIN cs AS cs_1 ON cs_1.id = bs_1.c_id
  WHERE EXISTS (SELECT 1
  FROM bs
  WHERE bs.id = `as`.b_id AND bs.x = ? AND (EXISTS (SELECT 1
  FROM cs
  WHERE cs.id = bs.c_id AND cs.y = ?)))
  

  嵌套子查询导致查询效率低下，大数据场景不适用。

• 使用.join()：关联字段未预加载

  select(A)
    .join(B, A.b_id == B.id)
    .join(C, B.c_id == C.id)
    .where(B.x == 0, C.y == 0)
  

  生成的SQL：

  SELECT `as`.id, `as`.b_id
  FROM `as` INNER JOIN bs ON `as`.b_id = bs.id INNER JOIN cs ON bs.c_id = cs.id
  WHERE bs.x = ? AND cs.y = ?
  

  查询结果不包含A.b和A.b.c字段，访问时会触发额外查询。

• 同时使用.join()和joinedload()：查询冗余

  select(A)
    .join(B, A.b_id == B.id)
    .join(C, B.c_id == C.id)
    .where(B.x == 0, C.y == 0)
    .options(joinedload(A.b).joinedload(B.c))
  

  生成的SQL：

  SELECT `as`.id, `as`.b_id, cs_1.id AS id_1, cs_1.y, bs_1.id AS id_2, bs_1.c_id, bs_1.x 
  FROM `as` INNER JOIN bs ON `as`.b_id = bs.id INNER JOIN cs ON bs.c_id = cs.id LEFT OUTER JOIN bs AS bs_1 ON bs_1.id = `as`.b_id LEFT OUTER JOIN cs AS cs_1 ON cs_1.id = bs_1.c_id
  WHERE bs.x = ? AND cs.y = ?
  

  出现重复的JOIN操作，造成冗余。

请问是否存在一种方式，既能像joinedload那样预加载A.b和A.b.c字段，又能生成类似.join()的高效SQL查询？

使用SQLAlchemy的contains_eager选项，它可以将显式JOIN的表与模型的关联属性绑定，复用JOIN结果来预加载关联字段，避免重复JOIN和子查询。

实现代码 #

from sqlalchemy import select, and_
from sqlalchemy.orm import contains_eager

query = (
    select(A)
    .join(A.b)
    .join(B.c)
    .where(and_(B.x == 0, C.y == 0))
    .options(
        contains_eager(A.b).contains_eager(B.c)
    )
)

生成的SQL #

SELECT `as`.id, `as`.b_id, bs.id AS id_1, bs.c_id, bs.x, cs.id AS id_2, cs.y
FROM `as` 
INNER JOIN bs ON `as`.b_id = bs.id 
INNER JOIN cs ON bs.c_id = cs.id
WHERE bs.x = ? AND cs.y = ?

说明 #

• contains_eager直接使用显式JOIN的表数据填充模型关联属性，无需额外LEFT JOIN，查询效率与直接使用.join()一致。
• A.b和A.b.c会被预加载，访问时不会触发延迟查询。
• 该方式同时满足高效查询和预加载需求，完美解决上述问题。

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