SqlAlchemy 2.0 中文文档（十二）（2）-阿里云开发者社区

SqlAlchemy 2.0 中文文档（十二）（1）https://developer.aliyun.com/article/1562925

自引用多对多关系

另见

本节记录了“邻接列表”模式的两表变体，该模式在邻接列表关系有所描述。务必查看子节自引用查询策略和配置自引用急切加载，这两者同样适用于此处讨论的映射模式。

多对多关系可以通过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参数确定我们希望如何连接到关联表。在上面的声明形式中，由于我们正在声明这些条件，因此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的 backref - 当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)

在上面的示例中，我们直接引用了命名为a、b、c、d的表，提供了relationship.secondary、relationship.primaryjoin和relationship.secondaryjoin这三个声明式样式的参数。从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

在上面的示例中，我们利用能够将多个表放入“secondary”容器的优势，以便我们可以跨多个表进行连接，同时保持对relationship()的“简化”，在这种情况下，“左”和“右”两侧都只有“一个”表；复杂性保持在中间。

警告

像上面的关系通常标记为viewonly=True，使用relationship.viewonly，应视为只读。虽然有时可以使类似上面的关系可写，但这通常很复杂且容易出错。

另请参阅

使用 viewonly 关系参数的注意事项 ## 别名类的关系

在前一节中，我们介绍了一种技术，其中我们使用relationship.secondary来在连接条件中放置额外的表。有一种复杂的连接情况，即使使用这种技术也不够；当我们试图从A连接到B，并在其中使用任意数量的C、D等，但A和B之间也直接有连接条件时。在这种情况下，仅仅使用一个复杂的relationship.primaryjoin条件可能难以表达从A到B的连接，因为中间表可能需要特殊处理，并且也不能用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)

上面，函数_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 
```## 使用窗口函数限制行关系
另一个与`AliasedClass`对象关系的有趣用例是在关系需要连接到任意形式的专门 SELECT 的情况下。一种情况是当需要使用窗口函数时，例如限制应返回多少行以供关系使用。下面的示例说明了一个非主映射器关系，它将为每个集合加载前十个项目：
```py
class A(Base):
    __tablename__ = "a"
    id = mapped_column(Integer, primary_key=True)
class B(Base):
    __tablename__ = "b"
    id = mapped_column(Integer, primary_key=True)
    a_id = mapped_column(ForeignKey("a.id"))
partition = select(
    B, func.row_number().over(order_by=B.id, partition_by=B.a_id).label("index")
).alias()
partitioned_b = aliased(B, partition)
A.partitioned_bs = relationship(
    partitioned_b, primaryjoin=and_(partitioned_b.a_id == A.id, partition.c.index < 10)
)

我们可以将上述partitioned_bs关系与大多数加载器策略一起使用，例如selectinload()：

for a1 in session.scalars(select(A).options(selectinload(A.partitioned_bs))):
    print(a1.partitioned_bs)  # <-- will be no more than ten objects

上面，“selectinload” 查询如下所示：

SELECT
  a_1.id  AS  a_1_id,  anon_1.id  AS  anon_1_id,  anon_1.a_id  AS  anon_1_a_id,
  anon_1.data  AS  anon_1_data,  anon_1.index  AS  anon_1_index
FROM  a  AS  a_1
JOIN  (
  SELECT  b.id  AS  id,  b.a_id  AS  a_id,  b.data  AS  data,
  row_number()  OVER  (PARTITION  BY  b.a_id  ORDER  BY  b.id)  AS  index
  FROM  b)  AS  anon_1
ON  anon_1.a_id  =  a_1.id  AND  anon_1.index  <  %(index_1)s
WHERE  a_1.id  IN  (  ...  primary  key  collection  ...)
ORDER  BY  a_1.id

上述情况下，对于“a”中的每个匹配主键，我们将按“b.id”排序获取前十个“bs”。通过在“a_id”上进行分区，我们确保每个“行号”都是相对于父“a_id”的局部的。

这样的映射通常也会包括从“A”到“B”的“普通”关系，用于持久性操作以及当需要每个“A”的完整“B”对象集合时。## 构建查询可用的属性

非常雄心勃勃的自定义连接条件可能无法直接持久化，并且在某些情况下甚至可能无法正确加载。要删除等式的持久性部分，请在relationship()上使用标志relationship.viewonly，将其建立为只读属性（写入到集合的数据将在刷新时被忽略）。然而，在极端情况下，考虑与Query一起使用普通的 Python 属性，如下所示：

class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    @property
    def addresses(self):
        return object_session(self).query(Address).with_parent(self).filter(...).all()

在其他情况下，描述符可以构建以利用现有的 Python 数据。有关使用描述符和混合体的更一般讨论，请参见使用描述符和混合体部分。

另见

使用描述符和混合体 ## 关于使用 viewonly 关系参数的注意事项

当应用于relationship()构造时，relationship.viewonly参数指示这个relationship()不会参与任何 ORM 工作单元操作，并且该属性不希望在其表示的集合的 Python 变异中参与。这意味着虽然只读关系可能引用一个可变的 Python 集合，如列表或集合，但对该列表或集合进行更改，如在映射实例上存在的那样，对 ORM 刷新过程没有影响。

要探索这种情况，请考虑这种映射：

from __future__ import annotations
import datetime
from sqlalchemy import and_
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
    pass
class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str | None]
    all_tasks: Mapped[list[Task]] = relationship()
    current_week_tasks: Mapped[list[Task]] = relationship(
        primaryjoin=lambda: and_(
            User.id == Task.user_account_id,
            # this expression works on PostgreSQL but may not be supported
            # by other database engines
            Task.task_date >= func.now() - datetime.timedelta(days=7),
        ),
        viewonly=True,
    )
class Task(Base):
    __tablename__ = "task"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    description: Mapped[str | None]
    task_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
    user: Mapped[User] = relationship(back_populates="current_week_tasks")

以下各节将注意这种配置的不同方面。

SqlAlchemy 2.0 中文文档（十二）（3）https://developer.aliyun.com/article/1562927

SqlAlchemy 2.0 中文文档（十二）（2）

自引用多对多关系

将别名类映射与类型和避免早期映射器配置集成

在查询中使用 AliasedClass 目标

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

SqlAlchemy 2.0 中文文档（十二）（2）

自引用多对多关系

将别名类映射与类型和避免早期映射器配置集成

在查询中使用 AliasedClass 目标

热门文章

最新文章

相关课程

相关电子书