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

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

包括反向引用的 Python 中的变异与 viewonly=True 不适用

上述映射将User.current_week_tasks只读关系作为Task.user属性的反向引用目标。这目前并未被 SQLAlchemy 的 ORM 配置过程标记，但这是一个配置错误。改变Task上的.user属性不会影响.current_week_tasks属性：

>>> u1 = User()
>>> t1 = Task(task_date=datetime.datetime.now())
>>> t1.user = u1
>>> u1.current_week_tasks
[]

这里还有另一个参数叫做relationship.sync_backrefs，可以在这里打开，以允许在这种情况下对.current_week_tasks进行变异，然而这并不被认为是最佳实践，对于一个只读关系，不应该依赖于 Python 中的变异。

在这种映射中，可以在User.all_tasks和Task.user之间配置反向引用，因为这两者都不是只读的，将正常同步。

除了禁用只读关系的反向引用变异问题外，Python 中对User.all_tasks集合的普通更改也不会反映在User.current_week_tasks集合中，直到更改已刷新到数据库。

总的来说，对于一个需要立即响应 Python 中的变异的自定义集合的用例，只读关系通常不合适。更好的方法是使用 SQLAlchemy 的混合属性功能，或者对于仅实例情况，使用 Python 的@property，其中可以实现一个根据当前 Python 实例生成的用户定义集合。要将我们的示例更改为这种方式工作，我们修复Task.user上的relationship.back_populates参数，引用User.all_tasks，然后演示一个简单的@property，将以立即User.all_tasks集合的形式提供结果：

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str | None]
    all_tasks: Mapped[list[Task]] = relationship(back_populates="user")
    @property
    def current_week_tasks(self) -> list[Task]:
        past_seven_days = datetime.datetime.now() - datetime.timedelta(days=7)
        return [t for t in self.all_tasks if t.task_date >= past_seven_days]
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="all_tasks")

使用在 Python 中每次动态计算的集合，我们保证始终有正确的答案，而无需使用数据库：

>>> u1 = User()
>>> t1 = Task(task_date=datetime.datetime.now())
>>> t1.user = u1
>>> u1.current_week_tasks
[<__main__.Task object at 0x7f3d699523c0>]

viewonly=True 的集合/属性直到过期才会重新查询

继续使用原始的 viewonly 属性，如果我们确实对 persistent 对象上的 User.all_tasks 集合进行更改，则在两个事情发生之后，viewonly 集合只能显示此更改的净结果。第一个是将更改刷新到 User.all_tasks 中，以便新数据在数据库中可用，至少在本地事务的范围内是如此。第二个是 User.current_week_tasks 属性被 expired 并通过对数据库进行新的 SQL 查询重新加载。

为了支持这一需求，最简单的流程是仅在主要是只读操作中使用 viewonly 关系。例如，如果我们从数据库中检索到一个新的 User，那么集合将是当前的：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f8711b906b0>]

当我们对 u1.all_tasks 进行修改时，如果想要在 u1.current_week_tasks 视图关系中看到这些更改，这些更改需要被刷新，并且 u1.current_week_tasks 属性需要过期，以便在下一次访问时进行惰性加载。最简单的方法是使用 Session.commit()，保持 Session.expire_on_commit 参数设置为其默认值 True：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     sess.commit()
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f8711b90ec0>, <__main__.Task object at 0x7f8711b90a10>]

上面，对 Session.commit() 的调用将更改刷新到了数据库中的 u1.all_tasks，然后使所有对象过期，因此当我们访问 u1.current_week_tasks 时，会发生 :term:惰性加载，从数据库中新鲜获取此属性的内容。

要拦截操作而不实际提交事务，需要先显式 expired 该属性。一种简单的方法是直接调用它。在下面的示例中，Session.flush() 将挂起的更改发送到数据库，然后使用 Session.expire() 来过期 u1.current_week_tasks 集合，以便在下一次访问时重新获取：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     sess.flush()
...     sess.expire(u1, ["current_week_tasks"])
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7fd95a4c8c50>, <__main__.Task object at 0x7fd95a4c8c80>]

事实上，我们可以跳过对Session.flush()的调用，假设一个保持Session.autoflush为其默认值True的Session，因为过期的current_week_tasks属性在过期后访问时将触发自动刷新：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     sess.expire(u1, ["current_week_tasks"])
...     print(u1.current_week_tasks)  # triggers autoflush before querying
[<__main__.Task object at 0x7fd95a4c8c50>, <__main__.Task object at 0x7fd95a4c8c80>]

继续使用上述方法进行更详细的处理，我们可以在相关的User.all_tasks集合发生变化时通过 event hooks 进行程序化过期。这是一种高级技术，应该首先检查更简单的架构，比如@property或坚持只读用例。在我们简单的示例中，这将被配置为：

from sqlalchemy import event, inspect
@event.listens_for(User.all_tasks, "append")
@event.listens_for(User.all_tasks, "remove")
@event.listens_for(User.all_tasks, "bulk_replace")
def _expire_User_current_week_tasks(target, value, initiator):
    inspect(target).session.expire(target, ["current_week_tasks"])

有了上述钩子，突变操作被拦截并导致User.current_week_tasks集合自动过期：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f66d093ccb0>, <__main__.Task object at 0x7f66d093cce0>]

上述使用的AttributeEvents事件钩子也会被 backref 突变触发，因此，使用上述钩子会拦截对Task.user的更改：

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     t1 = Task(task_date=datetime.datetime.now())
...     t1.user = u1
...     sess.add(t1)
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f3b0c070d10>, <__main__.Task object at 0x7f3b0c057d10>]
```## 处理多个连接路径
处理的最常见情况之一是两个表之间存在多个外键路径。
考虑一个包含对`Address`类的两个外键的`Customer`类：
```py
from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
    pass
class Customer(Base):
    __tablename__ = "customer"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)
    billing_address_id = mapped_column(Integer, ForeignKey("address.id"))
    shipping_address_id = mapped_column(Integer, ForeignKey("address.id"))
    billing_address = relationship("Address")
    shipping_address = relationship("Address")
class Address(Base):
    __tablename__ = "address"
    id = mapped_column(Integer, primary_key=True)
    street = mapped_column(String)
    city = mapped_column(String)
    state = mapped_column(String)
    zip = mapped_column(String)

上述映射，在我们尝试使用它时，会产生错误：

sqlalchemy.exc.AmbiguousForeignKeysError: Could not determine join
condition between parent/child tables on relationship
Customer.billing_address - there are multiple foreign key
paths linking the tables.  Specify the 'foreign_keys' argument,
providing a list of those columns which should be
counted as containing a foreign key reference to the parent table.

上述消息相当长。relationship()可能返回许多潜在消息，这些消息经过精心设计，以检测各种常见的配置问题；大多数消息都会建议需要解决模糊性或其他缺失信息的附加配置。

在这种情况下，消息希望我们为每个relationship()进行限定，指示每个外键列应该被考虑，并且适当的形式如下：

class Customer(Base):
    __tablename__ = "customer"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)
    billing_address_id = mapped_column(Integer, ForeignKey("address.id"))
    shipping_address_id = mapped_column(Integer, ForeignKey("address.id"))
    billing_address = relationship("Address", foreign_keys=[billing_address_id])
    shipping_address = relationship("Address", foreign_keys=[shipping_address_id])

在上面的例子中，我们指定了foreign_keys参数，它是一个Column或Column对象列表，指示要考虑的“外键”列，或者换句话说，包含指向父表的值的列。从Customer对象加载Customer.billing_address关系将使用billing_address_id中存在的值来标识要加载的Address行；类似地，shipping_address_id用于shipping_address关系。这两列的关联在持久化过程中也起着作用；刚插入的Address对象的新生成的主键将在刷新期间被复制到关联的Customer对象的适当外键列中。

在使用 Declarative 指定foreign_keys时，我们还可以使用字符串名称进行指定，但是重要的是，如果使用列表，列表是字符串的一部分：

billing_address = relationship("Address", foreign_keys="[Customer.billing_address_id]")

在这个具体的例子中，在任何情况下列表都是不必要的，因为我们只需要一个Column：

billing_address = relationship("Address", foreign_keys="Customer.billing_address_id")

警告

当作为 Python 可评估字符串传递时，relationship.foreign_keys 参数将使用 Python 的 eval() 函数进行解释。请勿将不受信任的输入传递给此字符串。详情请参阅关系参数的评估以了解有关relationship() 参数的声明性评估的详细信息。

指定备用连接条件

在构建连接时，relationship()的默认行为是将一侧的主键列的值等同于另一侧的外键引用列的值。我们可以使用relationship.primaryjoin参数来更改此标准为任何我们喜欢的内容，以及在使用“次要”表时，在使用relationship.secondaryjoin参数。

在下面的例子中，我们使用User类以及一个存储街道地址的Address类，我们创建了一个关系boston_addresses，它只会加载那些指定城市为“Boston”的Address对象：

from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship
class Base(DeclarativeBase):
    pass
class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)
    boston_addresses = relationship(
        "Address",
        primaryjoin="and_(User.id==Address.user_id, Address.city=='Boston')",
    )
class Address(Base):
    __tablename__ = "address"
    id = mapped_column(Integer, primary_key=True)
    user_id = mapped_column(Integer, ForeignKey("user.id"))
    street = mapped_column(String)
    city = mapped_column(String)
    state = mapped_column(String)
    zip = mapped_column(String)

在这个字符串的 SQL 表达式中，我们使用了 and_() 连接构造来建立两个不同的谓词，用于连接 User.id 和 Address.user_id 列，以及将 Address 中的行限制为只有 city='Boston'。当使用声明式时，类似 and_() 这样的基本 SQL 函数会自动在字符串 relationship() 参数的计算命名空间中可用。

警告

当作为 Python 可评估字符串传递时，relationship.primaryjoin 参数是使用 Python 的 eval() 函数解释的。不要将不受信任的输入传递给此字符串。有关声明式评估 relationship() 参数的详细信息，请参阅关系参数的评估。

我们在 relationship.primaryjoin 中使用的自定义条件通常只在 SQLAlchemy 渲染 SQL 以加载或表示此关系时才重要。也就是说，在执行每个属性的惰性加载的 SQL 语句中使用它，或者在查询时构造连接，例如通过 Select.join() 或通过急切的“连接”或“子查询”加载样式。当操作内存中的对象时，我们可以将任何我们想要的 Address 对象放入 boston_addresses 集合中，而不管 .city 属性的值是什么。这些对象将保留在集合中，直到属性过期并重新从应用条件的数据库中加载为止。当执行刷新时，boston_addresses 中的对象将被无条件地刷新，将主键 user.id 列的值分配到每行的持有外键 address.user_id 列。这里的 city 条件没有效果，因为刷新过程只关心将主键值同步到引用外键值。

创建自定义外键条件

主要连接条件的另一个元素是如何确定那些被认为是“外部”的列的。通常，一些 Column 对象的子集将指定 ForeignKey，或者是 ForeignKeyConstraint 的一部分，这与连接条件相关。relationship() 查看这个外键状态，以确定它应该如何为这个关系加载和持久化数据。然而，relationship.primaryjoin 参数可以用来创建一个不涉及任何“模式”级外键的连接条件。我们可以显式地结合 relationship.primaryjoin 以及 relationship.foreign_keys 和 relationship.remote_side 来建立这样一个连接。

下面，一个 HostEntry 类与自身连接，将字符串 content 列等同于 ip_address 列，这是一个名为 INET 的 PostgreSQL 类型。我们需要使用 cast() 来将连接的一侧转换为另一侧的类型：

from sqlalchemy import cast, String, Column, Integer
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import INET
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass
class HostEntry(Base):
    __tablename__ = "host_entry"
    id = mapped_column(Integer, primary_key=True)
    ip_address = mapped_column(INET)
    content = mapped_column(String(50))
    # relationship() using explicit foreign_keys, remote_side
    parent_host = relationship(
        "HostEntry",
        primaryjoin=ip_address == cast(content, INET),
        foreign_keys=content,
        remote_side=ip_address,
    )

上述关系将产生类似以下的连接：

SELECT  host_entry.id,  host_entry.ip_address,  host_entry.content
FROM  host_entry  JOIN  host_entry  AS  host_entry_1
ON  host_entry_1.ip_address  =  CAST(host_entry.content  AS  INET)

以上的另一种语法是在 relationship.primaryjoin 表达式内联使用 foreign() 和 remote() annotations。这种语法表示了 relationship() 通常自己应用于连接条件的注释，考虑到 relationship.foreign_keys 和 relationship.remote_side 参数。当存在明确的连接条件时，这些函数可能更简洁，并且还可以标记出“外部”或“远程”的确切列，而不管该列是否在多次声明或在复杂的 SQL 表达式中：

from sqlalchemy.orm import foreign, remote
class HostEntry(Base):
    __tablename__ = "host_entry"
    id = mapped_column(Integer, primary_key=True)
    ip_address = mapped_column(INET)
    content = mapped_column(String(50))
    # relationship() using explicit foreign() and remote() annotations
    # in lieu of separate arguments
    parent_host = relationship(
        "HostEntry",
        primaryjoin=remote(ip_address) == cast(foreign(content), INET),
    )

在连接条件中使用自定义运算符

另一个关系的用例是使用自定义运算符，比如 PostgreSQL 的“包含在内” << 运算符，当与诸如 INET 和 CIDR 这样的类型连接时。对于自定义布尔运算符，我们使用 Operators.bool_op() 函数：

inet_column.bool_op("<<")(cidr_column)

像上面的比较可以直接用于构造 relationship() 中的 relationship.primaryjoin：

class IPA(Base):
    __tablename__ = "ip_address"
    id = mapped_column(Integer, primary_key=True)
    v4address = mapped_column(INET)
    network = relationship(
        "Network",
        primaryjoin="IPA.v4address.bool_op('<<')(foreign(Network.v4representation))",
        viewonly=True,
    )
class Network(Base):
    __tablename__ = "network"
    id = mapped_column(Integer, primary_key=True)
    v4representation = mapped_column(CIDR)

上述，像这样的查询：

select(IPA).join(IPA.network)

将显示为：

SELECT  ip_address.id  AS  ip_address_id,  ip_address.v4address  AS  ip_address_v4address
FROM  ip_address  JOIN  network  ON  ip_address.v4address  <<  network.v4representation

基于 SQL 函数的自定义运算符

与 Operators.op.is_comparison 用例的变体是当我们不是使用运算符，而是使用 SQL 函数。这种用例的典型例子是 PostgreSQL PostGIS 函数，但任何解析为二进制条件的任何数据库上的 SQL 函数都可能适用。为适应这种用例，FunctionElement.as_comparison() 方法可以修改任何 SQL 函数，例如从 func 命名空间调用的函数，以指示 ORM 该函数生成了两个表达式的比较。下面的例子使用了 Geoalchemy2 库说明了这一点：

from geoalchemy2 import Geometry
from sqlalchemy import Column, Integer, func
from sqlalchemy.orm import relationship, foreign
class Polygon(Base):
    __tablename__ = "polygon"
    id = mapped_column(Integer, primary_key=True)
    geom = mapped_column(Geometry("POLYGON", srid=4326))
    points = relationship(
        "Point",
        primaryjoin="func.ST_Contains(foreign(Polygon.geom), Point.geom).as_comparison(1, 2)",
        viewonly=True,
    )
class Point(Base):
    __tablename__ = "point"
    id = mapped_column(Integer, primary_key=True)
    geom = mapped_column(Geometry("POINT", srid=4326))

上述，FunctionElement.as_comparison() 表示 func.ST_Contains() SQL 函数正在比较 Polygon.geom 和 Point.geom 表达式。foreign() 注释另外指出了在这种特定关系中扮演“外键”角色的列。

新版本 1.3 中新增了 FunctionElement.as_comparison()。

重叠的外键

很少见的情况可能会出现，即使用复合外键，以便单个列可能是通过外键约束引用的多个列的主题。

考虑一个（诚然复杂的）映射，例如Magazine对象，使用包括magazine_id的复合主键方案，分别由Writer对象和Article对象引用；然后，为了使Article也引用Writer，Article.magazine_id涉及到两个不同的关系；Article.magazine和Article.writer：

class Magazine(Base):
    __tablename__ = "magazine"
    id = mapped_column(Integer, primary_key=True)
class Article(Base):
    __tablename__ = "article"
    article_id = mapped_column(Integer)
    magazine_id = mapped_column(ForeignKey("magazine.id"))
    writer_id = mapped_column()
    magazine = relationship("Magazine")
    writer = relationship("Writer")
    __table_args__ = (
        PrimaryKeyConstraint("article_id", "magazine_id"),
        ForeignKeyConstraint(
            ["writer_id", "magazine_id"], ["writer.id", "writer.magazine_id"]
        ),
    )
class Writer(Base):
    __tablename__ = "writer"
    id = mapped_column(Integer, primary_key=True)
    magazine_id = mapped_column(ForeignKey("magazine.id"), primary_key=True)
    magazine = relationship("Magazine")

当上述映射被配置时，我们将看到此警告被发出：

SAWarning: relationship 'Article.writer' will copy column
writer.magazine_id to column article.magazine_id,
which conflicts with relationship(s): 'Article.magazine'
(copies magazine.id to article.magazine_id). Consider applying
viewonly=True to read-only relationships, or provide a primaryjoin
condition marking writable columns with the foreign() annotation.

这指的是Article.magazine_id是两个不同外键约束的主体；它直接引用Magazine.id作为源列，但在与Writer的复合键上下文中，也引用Writer.magazine_id作为源列。如果我们将Article与特定的Magazine关联起来，但然后将Article与另一个与不同Magazine关联的Writer关联起来，ORM 会非确定性地覆盖Article.magazine_id，悄悄地改变我们所引用的杂志；如果我们将Writer从Article中取消关联，它还可能尝试将 NULL 放入此列。警告让我们知道这种情况。

要解决这个问题，我们需要打破Article的行为，包括以下三个功能：

首先，Article根据仅在Article.magazine关系中持久化的数据来写入Article.magazine_id，即从Magazine.id复制的值。
Article可以代表在Article.writer关系中持久化的数据写入Article.writer_id，但只能写入Writer.id列；Writer.magazine_id列不应写入Article.magazine_id，因为它最终来自Magazine.id。
当加载Article.writer时，Article考虑了Article.magazine_id，即使在此关系中并不代表它。

要获取只有#1 和#2，我们可以将Article.writer_id指定为Article.writer的“外键”：

class Article(Base):
    # ...
    writer = relationship("Writer", foreign_keys="Article.writer_id")

然而，这会导致Article.writer在与Writer进行查询时不考虑Article.magazine_id：

SELECT  article.article_id  AS  article_article_id,
  article.magazine_id  AS  article_magazine_id,
  article.writer_id  AS  article_writer_id
FROM  article
JOIN  writer  ON  writer.id  =  article.writer_id

因此，要获取#1、#2 和#3 的所有内容，我们需要通过完全组合relationship.primaryjoin来表达连接条件，以及要写入的列，同时使用relationship.foreign_keys参数，或者更简洁地使用foreign()进行注释：

class Article(Base):
    # ...
    writer = relationship(
        "Writer",
        primaryjoin="and_(Writer.id == foreign(Article.writer_id), "
        "Writer.magazine_id == Article.magazine_id)",
    )

非关系比较 / 材料化路径

警告

本节详细介绍了一个实验性功能。

使用自定义表达式意味着我们可以生成不遵循通常的主键/外键模型的非正统连接条件。其中一个例子是材料化路径模式，我们在比较字符串以产生重叠路径标记时，以便生成树结构。

通过谨慎使用foreign()和remote()，我们可以构建一个有效地生成基本材料化路径系统的关系。基本上，当foreign()和remote()在相同的比较表达式一侧时，关系被认为是“一对多”；当它们在不同的一侧时，关系被认为是“多对一”。对于我们将在此处使用的比较，我们将处理集合，所以我们保持事物配置为“一对多”：

class Element(Base):
    __tablename__ = "element"
    path = mapped_column(String, primary_key=True)
    descendants = relationship(
        "Element",
        primaryjoin=remote(foreign(path)).like(path.concat("/%")),
        viewonly=True,
        order_by=path,
    )

上文中，如果给定具有"/foo/bar2"路径属性的Element对象，则我们寻找对Element.descendants的加载应如下所示：

SELECT  element.path  AS  element_path
FROM  element
WHERE  element.path  LIKE  ('/foo/bar2'  ||  '/%')  ORDER  BY  element.path

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

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

包括反向引用的 Python 中的变异与 viewonly=True 不适用

viewonly=True 的集合/属性直到过期才会重新查询

指定备用连接条件

创建自定义外键条件

在连接条件中使用自定义运算符

基于 SQL 函数的自定义运算符

重叠的外键

非关系比较 / 材料化路径

热门文章

最新文章

相关课程

相关电子书