SqlAlchemy 2.0 中文文档（十二）（3）https://developer.aliyun.com/article/1562927
自我引用多对多关系
另请参阅
本节记录了“邻接列表”模式的两个表变体，该模式在邻接列表关系中有所记录。务必查看自我引用查询模式的子部分自我引用查询策略和配置自我引用急加载，它们同样适用于此处讨论的映射模式。
多对多关系可以通过relationship.primaryjoin和/或relationship.secondaryjoin中的一个或两个进行自定义 - 后者对于使用relationship.secondary参数指定多对多引用的关系非常重要。涉及使用relationship.primaryjoin和relationship.secondaryjoin的常见情况是在从一个类到自身建立多对多关系时，如下所示：
from typing import List from sqlalchemy import Integer, ForeignKey, Column, Table from sqlalchemy.orm import DeclarativeBase, Mapped from sqlalchemy.orm import mapped_column, relationship class Base(DeclarativeBase): pass node_to_node = Table( "node_to_node", Base.metadata, Column("left_node_id", Integer, ForeignKey("node.id"), primary_key=True), Column("right_node_id", Integer, ForeignKey("node.id"), primary_key=True), ) class Node(Base): __tablename__ = "node" id: Mapped[int] = mapped_column(primary_key=True) label: Mapped[str] right_nodes: Mapped[List["Node"]] = relationship( "Node", secondary=node_to_node, primaryjoin=id == node_to_node.c.left_node_id, secondaryjoin=id == node_to_node.c.right_node_id, back_populates="left_nodes", ) left_nodes: Mapped[List["Node"]] = relationship( "Node", secondary=node_to_node, primaryjoin=id == node_to_node.c.right_node_id, secondaryjoin=id == node_to_node.c.left_node_id, back_populates="right_nodes", )
在上述情况下，SQLAlchemy 无法自动知道哪些列应该连接到right_nodes和left_nodes关系的哪些列上。relationship.primaryjoin和relationship.secondaryjoin参数建立了我们希望如何加入关联表的方式。在上面的声明形式中，由于我们在对应于Node类的 Python 块中声明了这些条件，因此id变量直接作为我们希望与之连接的Column对象是可用的。
或者，我们可以使用字符串定义relationship.primaryjoin和relationship.secondaryjoin参数，在我们的配置中可能还没有Node.id列对象可用，或者node_to_node表可能还不可用的情况下非常适用。当在声明字符串中引用普通的Table对象时，我们使用表的字符串名称，就像在MetaData中一样：
class Node(Base): __tablename__ = "node" id = mapped_column(Integer, primary_key=True) label = mapped_column(String) right_nodes = relationship( "Node", secondary="node_to_node", primaryjoin="Node.id==node_to_node.c.left_node_id", secondaryjoin="Node.id==node_to_node.c.right_node_id", backref="left_nodes", )
警告
当作为 Python 可评估字符串传递时，relationship.primaryjoin和relationship.secondaryjoin参数使用 Python 的eval()函数解释。不要将不受信任的输入传递给这些字符串。有关声明性relationship()参数的评估详细信息，请参阅关系参数的评估。
在这里的经典映射情况中，node_to_node可以连接到node.c.id：
from sqlalchemy import Integer, ForeignKey, String, Column, Table, MetaData from sqlalchemy.orm import relationship, registry metadata_obj = MetaData() mapper_registry = registry() node_to_node = Table( "node_to_node", metadata_obj, Column("left_node_id", Integer, ForeignKey("node.id"), primary_key=True), Column("right_node_id", Integer, ForeignKey("node.id"), primary_key=True), ) node = Table( "node", metadata_obj, Column("id", Integer, primary_key=True), Column("label", String), ) class Node: pass mapper_registry.map_imperatively( Node, node, properties={ "right_nodes": relationship( Node, secondary=node_to_node, primaryjoin=node.c.id == node_to_node.c.left_node_id, secondaryjoin=node.c.id == node_to_node.c.right_node_id, backref="left_nodes", ) }, )
请注意，在这两个示例中，relationship.backref 关键字都指定了一个 left_nodes 回引 - 当relationship()在反向创建第二个关系时，它足够聪明地反转了relationship.primaryjoin 和 relationship.secondaryjoin 参数。
另请参阅
- 邻接列表关系 - 单表版本
- 自引用查询策略 - 使用自引用映射查询的技巧
- 配置自引用预加载 - 使用自引用映射预加载的技巧
复合“次要”连接
注意
本节涵盖了一些在某种程度上受 SQLAlchemy 支持的边缘案例，但建议尽可能在可能的情况下通过使用合理的关系布局和/或 Python 内的属性来解决类似问题。
有时，当一个人试图在两个表之间建立一个relationship()时，需要涉及更多的表，而不仅仅是两个或三个表。这是一个relationship()的领域，在这个领域，人们试图推动可能性的边界，并且通常对许多这些奇特用例的最终解决方案需要在 SQLAlchemy 邮件列表上讨论。
在较新版本的 SQLAlchemy 中，relationship.secondary 参数可用于某些情况，以提供由多个表组成的复合目标。以下是这种连接条件的示例（至少需要版本 0.9.2 才能正常运行）：
class A(Base): __tablename__ = "a" id = mapped_column(Integer, primary_key=True) b_id = mapped_column(ForeignKey("b.id")) d = relationship( "D", secondary="join(B, D, B.d_id == D.id).join(C, C.d_id == D.id)", primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)", secondaryjoin="D.id == B.d_id", uselist=False, viewonly=True, ) class B(Base): __tablename__ = "b" id = mapped_column(Integer, primary_key=True) d_id = mapped_column(ForeignKey("d.id")) class C(Base): __tablename__ = "c" id = mapped_column(Integer, primary_key=True) a_id = mapped_column(ForeignKey("a.id")) d_id = mapped_column(ForeignKey("d.id")) class D(Base): __tablename__ = "d" id = mapped_column(Integer, primary_key=True)
在上面的示例中，我们提供了relationship.secondary、relationship.primaryjoin 和relationship.secondaryjoin 这三个参数，在声明样式中直接引用了命名表 a、b、c、d。从 A 到 D 的查询如下：
sess.scalars(select(A).join(A.d)).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN ( b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id
在上面的示例中，我们利用能够将多个表填入“次要”容器的优势，这样我们就可以跨多个表进行连接，同时保持对relationship()的“简单”使用，因为“左”和“右”两侧只有“一个”表；复杂性被保留在中间。
警告
类似上述的关系通常标记为viewonly=True，使用relationship.viewonly，应当被视为只读。虽然有时可以使类似上述的关系可写，但这通常是复杂且容易出错的。
另请参阅
关于使用只读关系参数的注意事项
别名类的关系
在前一节中，我们说明了一种技术，在这种技术中，我们使用了relationship.secondary来将额外的表放置在连接条件中。有一种复杂的连接情况，即使使用这种技术也不足够；当我们试图从A连接到B时，中间可能会使用任意数量的C、D等，但是在A和B之间也有直接的连接条件。在这种情况下，仅使用复杂的relationship.primaryjoin条件可能难以表达，因为中间表可能需要特殊处理，而且也不能使用relationship.secondary对象来表达，因为A->secondary->B模式不支持A和B之间的任何引用。当出现这种极其高级的情况时，我们可以采用创建第二个映射作为关系的目标。这就是我们使用AliasedClass来创建一个包含我们所需的所有额外表的类的映射。为了将这个映射作为我们类的“替代”映射生成，我们使用aliased()函数来生成新的结构，然后针对该对象使用relationship()，就像它是一个普通的映射类一样。
下面演示了从 A 到 B 的简单连接的 relationship()，但是主连接条件增加了另外两个实体 C 和 D，这两个实体必须同时与 A 和 B 中的行对应起来：
class A(Base): __tablename__ = "a" id = mapped_column(Integer, primary_key=True) b_id = mapped_column(ForeignKey("b.id")) class B(Base): __tablename__ = "b" id = mapped_column(Integer, primary_key=True) class C(Base): __tablename__ = "c" id = mapped_column(Integer, primary_key=True) a_id = mapped_column(ForeignKey("a.id")) some_c_value = mapped_column(String) class D(Base): __tablename__ = "d" id = mapped_column(Integer, primary_key=True) c_id = mapped_column(ForeignKey("c.id")) b_id = mapped_column(ForeignKey("b.id")) some_d_value = mapped_column(String) # 1\. set up the join() as a variable, so we can refer # to it in the mapping multiple times. j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id) # 2\. Create an AliasedClass to B B_viacd = aliased(B, j, flat=True) A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
使用上述映射，简单连接如下所示：
sess.scalars(select(A).join(A.b)).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) ON a.b_id = b.id
将别名类映射与类型化结合，并避免早期映射器配置
aliased() 构造对映射类的创建强制执行 configure_mappers() 步骤，该步骤将解析所有当前类及其关系。如果当前映射所需的不相关映射类尚未声明，或者如果关系本身的配置需要访问尚未声明的类，则可能会出现问题。此外，当关系在前面声明时，SQLAlchemy 的声明模式与 Python 类型化的协作效果最佳。
为了组织关系的构建以解决这些问题，可以使用配置级别的事件钩子，如 MapperEvents.before_mapper_configured()，该钩子将仅在所有映射准备好配置时调用配置代码：
from sqlalchemy import event class A(Base): __tablename__ = "a" id = mapped_column(Integer, primary_key=True) b_id = mapped_column(ForeignKey("b.id")) @event.listens_for(A, "before_mapper_configured") def _configure_ab_relationship(mapper, cls): # do the above configuration in a configuration hook j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id) B_viacd = aliased(B, j, flat=True) A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
在上面的示例中，当请求完全配置的 A 版本时，函数 _configure_ab_relationship() 将被调用，此时类 B、D 和 C 将可用。
对于与内联类型化集成的方法，可以使用类似的技术来有效地生成别名类的“单例”创建模式，其中它作为全局变量进行延迟初始化，然后可以在关系内联中使用它：
from typing import Any B_viacd: Any = None b_viacd_join: Any = None class A(Base): __tablename__ = "a" id: Mapped[int] = mapped_column(primary_key=True) b_id: Mapped[int] = mapped_column(ForeignKey("b.id")) # 1\. the relationship can be declared using lambdas, allowing it to resolve # to targets that are late-configured b: Mapped[B] = relationship( lambda: B_viacd, primaryjoin=lambda: A.b_id == b_viacd_join.c.b_id ) # 2\. configure the targets of the relationship using a before_mapper_configured # hook. @event.listens_for(A, "before_mapper_configured") def _configure_ab_relationship(mapper, cls): # 3\. set up the join() and AliasedClass as globals from within # the configuration hook. global B_viacd, b_viacd_join b_viacd_join = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id) B_viacd = aliased(B, b_viacd_join, flat=True)
在查询中使用别名类目标
在前面的示例中，A.b 关系将 B_viacd 实体作为目标，而 不是 直接的 B 类。要添加涉及 A.b 关系的附加条件，通常需要直接引用 B_viacd 而不是使用 B，特别是在将 A.b 的目标实体转换为别名或子查询的情况下。下面演示了使用子查询而不是连接的相同关系：
subq = select(B).join(D, D.b_id == B.id).join(C, C.id == D.c_id).subquery() B_viacd_subquery = aliased(B, subq) A.b = relationship(B_viacd_subquery, primaryjoin=A.b_id == subq.c.id)
使用上述 A.b 关系的查询将呈现一个子查询：
sess.scalars(select(A).join(A.b)).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id
如果我们想要根据 A.b 连接添加额外的条件，必须以 B_viacd_subquery 而不是直接以 B 的形式添加：
sess.scalars( select(A) .join(A.b) .where(B_viacd_subquery.some_b_column == "some b") .order_by(B_viacd_subquery.id) ).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id WHERE anon_1.some_b_column = ? ORDER BY anon_1.id
将别名类映射与类型化结合，并避免早期映射器配置
对映射类创建 aliased() 构造会强制进行 configure_mappers() 步骤，这将解析所有当前的类及其关系。如果当前映射需要未声明的不相关的映射类，或者如果关系的配置本身需要访问尚未声明的类，则可能会出现问题。另外，当关系提前声明时，SQLAlchemy 的声明式模式与 Python 类型的工作方式最有效。
要组织关系构建以处理这些问题，可以使用一个配置级别的事件钩子，比如 MapperEvents.before_mapper_configured()，它只有在所有映射都准备好配置时才会调用配置代码：
from sqlalchemy import event class A(Base): __tablename__ = "a" id = mapped_column(Integer, primary_key=True) b_id = mapped_column(ForeignKey("b.id")) @event.listens_for(A, "before_mapper_configured") def _configure_ab_relationship(mapper, cls): # do the above configuration in a configuration hook j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id) B_viacd = aliased(B, j, flat=True) A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
上述函数 _configure_ab_relationship() 只有在请求一个完全配置好的 A 版本时才会被调用，在这时 B、D 和 C 类会被加载。
对于与内联类型集成的方法，可以使用类似的技术有效地为别名类生成“单例”创建模式，在其中作为全局变量进行延迟初始化，然后可以在关系中使用：
from typing import Any B_viacd: Any = None b_viacd_join: Any = None class A(Base): __tablename__ = "a" id: Mapped[int] = mapped_column(primary_key=True) b_id: Mapped[int] = mapped_column(ForeignKey("b.id")) # 1\. the relationship can be declared using lambdas, allowing it to resolve # to targets that are late-configured b: Mapped[B] = relationship( lambda: B_viacd, primaryjoin=lambda: A.b_id == b_viacd_join.c.b_id ) # 2\. configure the targets of the relationship using a before_mapper_configured # hook. @event.listens_for(A, "before_mapper_configured") def _configure_ab_relationship(mapper, cls): # 3\. set up the join() and AliasedClass as globals from within # the configuration hook. global B_viacd, b_viacd_join b_viacd_join = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id) B_viacd = aliased(B, b_viacd_join, flat=True)
在查询中使用 AliasedClass 目标
在前面的示例中，A.b 关系将 B_viacd 实体作为目标，而不是直接使用 B 类。要添加涉及 A.b 关系的额外条件，通常需要直接引用 B_viacd 而不是使用 B，特别是在目标实体 A.b 需要转换为别名或子查询的情况下。以下是使用子查询而不是连接的相同关系的示例：
subq = select(B).join(D, D.b_id == B.id).join(C, C.id == D.c_id).subquery() B_viacd_subquery = aliased(B, subq) A.b = relationship(B_viacd_subquery, primaryjoin=A.b_id == subq.c.id)
使用上述 A.b 关系的查询将呈现一个子查询：
sess.scalars(select(A).join(A.b)).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id
如果我们想要根据 A.b 连接添加额外的条件，必须以 B_viacd_subquery 而不是直接使用 B：
sess.scalars( select(A) .join(A.b) .where(B_viacd_subquery.some_b_column == "some b") .order_by(B_viacd_subquery.id) ).all() SELECT a.id AS a_id, a.b_id AS a_b_id FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id WHERE anon_1.some_b_column = ? ORDER BY anon_1.id
SqlAlchemy 2.0 中文文档（十二）（5）https://developer.aliyun.com/article/1562929
