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 目标

热门文章

最新文章

相关课程

相关电子书