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

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

写入仅集合 - API 文档

对象名称	描述
WriteOnlyCollection	写入仅集合，可将更改同步到属性事件系统中。
WriteOnlyMapped	代表“仅写”关系的 ORM 映射属性类型。

class sqlalchemy.orm.WriteOnlyCollection

写入仅集合，可将更改同步到属性事件系统中。

WriteOnlyCollection在映射中使用"write_only"延迟加载策略与relationship()一起使用。有关此配置的背景，请参阅仅写关系。

版本 2.0 中的新功能。

请参阅

仅写关系

成员

add(), add_all(), delete(), insert(), remove(), select(), update()

类签名

class sqlalchemy.orm.WriteOnlyCollection (sqlalchemy.orm.writeonly.AbstractCollectionWriter)

method add(item: _T) → None

向此WriteOnlyCollection添加一个项目。

下一次刷新时，给定的项目将以父实例的集合的形式持久化到数据库中。

method add_all(iterator: Iterable[_T]) → None

向此WriteOnlyCollection添加一组项目。

下一次刷新时，给定的项目将以父实例的集合的形式持久化到数据库中。

method delete() → Delete

生成一个 Delete，将引用以这个实例本地的WriteOnlyCollection。

method insert() → Inser

对于一对多集合，产生一个 Insert，该插入将以此实例本地 WriteOnlyCollection 为条件插入新的行。

此构造仅支持不包含 relationship.secondary 参数的 Relationship。对于引用到多对多表的关系，请使用普通的批量插入技术来产生新对象，然后使用 AbstractCollectionWriter.add_all() 将其与集合关联起来。

method remove(item: _T) → None

从此 WriteOnlyCollection 中移除一个项目。

给定的项目将在下次刷新时从父实例的集合中移除。

method select() → Select[Tuple[_T]]

产生一个 Select 构造，表示此实例本地 WriteOnlyCollection 中的行。

method update() → Update

产生一个 Update，该更新将参考以此实例本地为条件的行的 WriteOnlyCollection。

class sqlalchemy.orm.WriteOnlyMapped

表示“只写”关系的 ORM 映射属性类型。

WriteOnlyMapped 类型注释可以在带注释的声明性表映射中使用，以指示特定的 relationship() 应使用 lazy="write_only" 加载策略。

例如：

class User(Base):
 __tablename__ = "user"
 id: Mapped[int] = mapped_column(primary_key=True)
 addresses: WriteOnlyMapped[Address] = relationship(
 cascade="all,delete-orphan"
 )

参见章节只写关系了解背景。

新版本 2.0 中新增。

另请参阅

只写关系 - 完整的背景

DynamicMapped - 包含旧的 Query 支持

类签名

类 sqlalchemy.orm.WriteOnlyMapped (sqlalchemy.orm.base._MappedAnnotationBase)

创建和持久化新的只写集合

仅写集合允许直接将集合整体分配为仅用于瞬态或待处理对象。根据我们上面的映射，这表示我们可以创建一个新的 Account 对象，其中包含要添加到Session中的一系列 AccountTransaction 对象。任何 Python 可迭代对象都可以用作要开始的对象的源，下面我们使用 Python list：

>>> new_account = Account(
...     identifier="account_01",
...     account_transactions=[
...         AccountTransaction(description="initial deposit", amount=Decimal("500.00")),
...         AccountTransaction(description="transfer", amount=Decimal("1000.00")),
...         AccountTransaction(description="withdrawal", amount=Decimal("-29.50")),
...     ],
... )
>>> with Session(engine) as session:
...     session.add(new_account)
...     session.commit()
BEGIN  (implicit)
INSERT  INTO  account  (identifier)  VALUES  (?)
[...]  ('account_01',)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)
VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)  RETURNING  id,  timestamp
[...  (insertmanyvalues)  1/3  (ordered;  batch  not  supported)]  (1,  'initial deposit',  500.0)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)
VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)  RETURNING  id,  timestamp
[insertmanyvalues  2/3  (ordered;  batch  not  supported)]  (1,  'transfer',  1000.0)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)
VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)  RETURNING  id,  timestamp
[insertmanyvalues  3/3  (ordered;  batch  not  supported)]  (1,  'withdrawal',  -29.5)
COMMIT

一旦对象被数据库持久化（即处于持久化或分离状态），集合就有能力扩展新项目以及个别项目的能力被移除。但是，集合可能不再被重新分配为完整替换集合，因为这样的操作要求以前的集合完全加载到内存中，以便将旧条目与新条目进行对比：

>>> new_account.account_transactions = [
...     AccountTransaction(description="some transaction", amount=Decimal("10.00"))
... ]
Traceback (most recent call last):
...
sqlalchemy.exc.InvalidRequestError: Collection "Account.account_transactions" does not
support implicit iteration; collection replacement operations can't be used

向现有集合添加新项目

对于持久对象的仅写集合，使用 unit of work 过程对集合进行修改只能使用WriteOnlyCollection.add()、WriteOnlyCollection.add_all()和WriteOnlyCollection.remove()方法：

>>> from sqlalchemy import select
>>> session = Session(engine, expire_on_commit=False)
>>> existing_account = session.scalar(select(Account).filter_by(identifier="account_01"))
BEGIN  (implicit)
SELECT  account.id,  account.identifier
FROM  account
WHERE  account.identifier  =  ?
[...]  ('account_01',)
>>> existing_account.account_transactions.add_all(
...     [
...         AccountTransaction(description="paycheck", amount=Decimal("2000.00")),
...         AccountTransaction(description="rent", amount=Decimal("-800.00")),
...     ]
... )
>>> session.commit()
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)
VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)  RETURNING  id,  timestamp
[...  (insertmanyvalues)  1/2  (ordered;  batch  not  supported)]  (1,  'paycheck',  2000.0)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)
VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)  RETURNING  id,  timestamp
[insertmanyvalues  2/2  (ordered;  batch  not  supported)]  (1,  'rent',  -800.0)
COMMIT

上述添加的项目在Session中被保留在待处理队列中，直到下一个 flush，此时它们被插入到数据库中，假设添加的对象之前是瞬态的。

查询项目

WriteOnlyCollection不会在任何时候存储对集合当前内容的引用，也不会有任何直接发出 SELECT 到数据库以加载它们的行为；其覆盖的假设是集合可能包含许多千万个或数百万个行，并且不应作为任何其他操作的副作用完全加载到内存中。

相反，WriteOnlyCollection 包括生成 SQL 的辅助工具，如 WriteOnlyCollection.select()，它将生成一个预先配置了正确的 WHERE / FROM 条件的 Select 构造，然后可以进一步修改以选择所需的任何行范围，还可以使用服务器端游标进行调用，以便以内存有效的方式遍历整个集合。

下面说明了生成的语句。请注意，示例映射中的 relationship.order_by 参数指示了 ORDER BY 条件；如果未配置该参数，则该条件将被省略：

>>> print(existing_account.account_transactions.select())
SELECT  account_transaction.id,  account_transaction.account_id,  account_transaction.description,
account_transaction.amount,  account_transaction.timestamp
FROM  account_transaction
WHERE  :param_1  =  account_transaction.account_id  ORDER  BY  account_transaction.timestamp

我们可以使用此 Select 构造以及 Session 来查询 AccountTransaction 对象，最简单的方法是使用 Session.scalars() 方法，该方法将返回一个直接产生 ORM 对象的 Result。通常情况下，但不是必需的，会进一步修改 Select 以限制返回的记录；在下面的示例中，添加了额外的 WHERE 条件以仅加载 “借方” 账户交易，并添加了 “LIMIT 10” 以仅检索前十行：

>>> account_transactions = session.scalars(
...     existing_account.account_transactions.select()
...     .where(AccountTransaction.amount < 0)
...     .limit(10)
... ).all()
BEGIN  (implicit)
SELECT  account_transaction.id,  account_transaction.account_id,  account_transaction.description,
account_transaction.amount,  account_transaction.timestamp
FROM  account_transaction
WHERE  ?  =  account_transaction.account_id  AND  account_transaction.amount  <  ?
ORDER  BY  account_transaction.timestamp  LIMIT  ?  OFFSET  ?
[...]  (1,  0,  10,  0)
>>> print(account_transactions)
[AccountTransaction(amount=-29.50, timestamp='...'), AccountTransaction(amount=-800.00, timestamp='...')]

删除项目

在当前 Session 中针对持久状态的单个加载项目可以使用 WriteOnlyCollection.remove() 方法标记为从集合中删除。当操作进行时，刷新过程将隐式考虑对象已经是集合的一部分。下面的示例说明了如何删除单个 AccountTransaction 项目，根据级联设置，这将导致删除该行：

>>> existing_transaction = account_transactions[0]
>>> existing_account.account_transactions.remove(existing_transaction)
>>> session.commit()
DELETE  FROM  account_transaction  WHERE  account_transaction.id  =  ?
[...]  (3,)
COMMIT

与任何 ORM 映射的集合一样，对象的移除可以选择将对象与集合解除关联，同时保留对象在数据库中，或者可以基于 relationship() 的 delete-orphan 配置发出其行的 DELETE。

在不删除的情况下移除集合涉及将外键列设置为 NULL（对于一对多关系）或删除相应的关联行（对于多对多关系）。

批量插入新项目

WriteOnlyCollection 可以生成诸如 Insert 对象之类的 DML 构造，这些构造可以在 ORM 上下文中用于产生批量插入行为。参见 ORM 批量插入语句章节了解 ORM 批量插入的概述。

一对多集合

仅适用于常规的一对多集合，WriteOnlyCollection.insert() 方法将产生一个预先设定了与父对象相对应的 VALUES 条件的 Insert 构造。由于这个 VALUES 条件完全针对相关表，该语句可用于插入新行，这些新行将同时成为相关集合中的新记录：

>>> session.execute(
...     existing_account.account_transactions.insert(),
...     [
...         {"description": "transaction 1", "amount": Decimal("47.50")},
...         {"description": "transaction 2", "amount": Decimal("-501.25")},
...         {"description": "transaction 3", "amount": Decimal("1800.00")},
...         {"description": "transaction 4", "amount": Decimal("-300.00")},
...     ],
... )
BEGIN  (implicit)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)  VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)
[...]  [(1,  'transaction 1',  47.5),  (1,  'transaction 2',  -501.25),  (1,  'transaction 3',  1800.0),  (1,  'transaction 4',  -300.0)]
<...>
>>> session.commit()
COMMIT

另请参阅

ORM 批量插入语句 - 在 ORM 查询指南中

一对多 - 在基本关系模式中

多对多集合

对于多对多集合，两个类之间的关系涉及使用 relationship.secondary 参数配置的第三个表的情况。要使用 WriteOnlyCollection 批量插入此类型的集合中的行，新记录可能首先单独进行批量插入，然后使用 RETURNING 检索，然后将这些记录传递给 WriteOnlyCollection.add_all() 方法，在这个过程中，工作单元将会将它们作为集合的一部分持久化。

假设一个类BankAudit使用多对多表引用了许多AccountTransaction记录：

>>> from sqlalchemy import Table, Column
>>> audit_to_transaction = Table(
...     "audit_transaction",
...     Base.metadata,
...     Column("audit_id", ForeignKey("audit.id", ondelete="CASCADE"), primary_key=True),
...     Column(
...         "transaction_id",
...         ForeignKey("account_transaction.id", ondelete="CASCADE"),
...         primary_key=True,
...     ),
... )
>>> class BankAudit(Base):
...     __tablename__ = "audit"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     account_transactions: WriteOnlyMapped["AccountTransaction"] = relationship(
...         secondary=audit_to_transaction, passive_deletes=True
...     )

为了说明这两个操作，我们使用批量插入添加更多AccountTransaction对象，我们通过将returning(AccountTransaction)添加到批量 INSERT 语句中使用 RETURNING 检索（注意我们也可以使用现有的AccountTransaction对象）：

>>> new_transactions = session.scalars(
...     existing_account.account_transactions.insert().returning(AccountTransaction),
...     [
...         {"description": "odd trans 1", "amount": Decimal("50000.00")},
...         {"description": "odd trans 2", "amount": Decimal("25000.00")},
...         {"description": "odd trans 3", "amount": Decimal("45.00")},
...     ],
... ).all()
BEGIN  (implicit)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)  VALUES
(?,  ?,  ?,  CURRENT_TIMESTAMP),  (?,  ?,  ?,  CURRENT_TIMESTAMP),  (?,  ?,  ?,  CURRENT_TIMESTAMP)
RETURNING  id,  account_id,  description,  amount,  timestamp
[...]  (1,  'odd trans 1',  50000.0,  1,  'odd trans 2',  25000.0,  1,  'odd trans 3',  45.0)

有一组准备好的AccountTransaction对象列表，使用WriteOnlyCollection.add_all()方法可以一次性将许多行与新的BankAudit对象关联起来：

>>> bank_audit = BankAudit()
>>> session.add(bank_audit)
>>> bank_audit.account_transactions.add_all(new_transactions)
>>> session.commit()
INSERT  INTO  audit  DEFAULT  VALUES
[...]  ()
INSERT  INTO  audit_transaction  (audit_id,  transaction_id)  VALUES  (?,  ?)
[...]  [(1,  10),  (1,  11),  (1,  12)]
COMMIT

另请参阅

ORM 批量 INSERT 语句 - 在 ORM 查询指南中

多对多 - 在基本关系模式中

一对多集合

对于普通的一对多集合，WriteOnlyCollection.insert()方法将生成一个与父对象对应的 VALUES 条件预设的Insert构造。由于这个 VALUES 条件完全针对相关表，该语句可用于插入新行，同时这些新行也将成为相关集合中的新记录：

>>> session.execute(
...     existing_account.account_transactions.insert(),
...     [
...         {"description": "transaction 1", "amount": Decimal("47.50")},
...         {"description": "transaction 2", "amount": Decimal("-501.25")},
...         {"description": "transaction 3", "amount": Decimal("1800.00")},
...         {"description": "transaction 4", "amount": Decimal("-300.00")},
...     ],
... )
BEGIN  (implicit)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)  VALUES  (?,  ?,  ?,  CURRENT_TIMESTAMP)
[...]  [(1,  'transaction 1',  47.5),  (1,  'transaction 2',  -501.25),  (1,  'transaction 3',  1800.0),  (1,  'transaction 4',  -300.0)]
<...>
>>> session.commit()
COMMIT

另请参阅

ORM 批量 INSERT 语句 - 在 ORM 查询指南中

一对多 - 在基本关系模式中

多对多集合

对于多对多集合，两个类之间的关系涉及使用relationship的relationship.secondary参数配置的第三个表。要使用WriteOnlyCollection批量插入此类型的集合中的行，新记录可能首先被单独批量插入，然后使用 RETURNING 检索，并将这些记录传递给WriteOnlyCollection.add_all()方法，其中工作单元过程将继续将它们持久化为集合的一部分。

假设一个类BankAudit使用多对多表引用了许多AccountTransaction记录：

>>> from sqlalchemy import Table, Column
>>> audit_to_transaction = Table(
...     "audit_transaction",
...     Base.metadata,
...     Column("audit_id", ForeignKey("audit.id", ondelete="CASCADE"), primary_key=True),
...     Column(
...         "transaction_id",
...         ForeignKey("account_transaction.id", ondelete="CASCADE"),
...         primary_key=True,
...     ),
... )
>>> class BankAudit(Base):
...     __tablename__ = "audit"
...     id: Mapped[int] = mapped_column(primary_key=True)
...     account_transactions: WriteOnlyMapped["AccountTransaction"] = relationship(
...         secondary=audit_to_transaction, passive_deletes=True
...     )

为了说明这两个操作，我们使用批量插入添加更多的AccountTransaction对象，我们通过在批量插入语句中添加returning(AccountTransaction)来检索这些对象（请注意，我们也可以轻松地使用现有的AccountTransaction对象）：

>>> new_transactions = session.scalars(
...     existing_account.account_transactions.insert().returning(AccountTransaction),
...     [
...         {"description": "odd trans 1", "amount": Decimal("50000.00")},
...         {"description": "odd trans 2", "amount": Decimal("25000.00")},
...         {"description": "odd trans 3", "amount": Decimal("45.00")},
...     ],
... ).all()
BEGIN  (implicit)
INSERT  INTO  account_transaction  (account_id,  description,  amount,  timestamp)  VALUES
(?,  ?,  ?,  CURRENT_TIMESTAMP),  (?,  ?,  ?,  CURRENT_TIMESTAMP),  (?,  ?,  ?,  CURRENT_TIMESTAMP)
RETURNING  id,  account_id,  description,  amount,  timestamp
[...]  (1,  'odd trans 1',  50000.0,  1,  'odd trans 2',  25000.0,  1,  'odd trans 3',  45.0)

准备好一个AccountTransaction对象列表后，使用WriteOnlyCollection.add_all()方法一次将许多行与新的BankAudit对象关联起来：

>>> bank_audit = BankAudit()
>>> session.add(bank_audit)
>>> bank_audit.account_transactions.add_all(new_transactions)
>>> session.commit()
INSERT  INTO  audit  DEFAULT  VALUES
[...]  ()
INSERT  INTO  audit_transaction  (audit_id,  transaction_id)  VALUES  (?,  ?)
[...]  [(1,  10),  (1,  11),  (1,  12)]
COMMIT

请参见

ORM 批量插入语句 - 在 ORM 查询指南

多对多 - 在基本关系模式

SqlAlchemy 2.0 中文文档（十三）（5）https://developer.aliyun.com/article/1562964

SqlAlchemy 2.0 中文文档（十三）（4）

写入仅集合 - API 文档

创建和持久化新的只写集合

向现有集合添加新项目

查询项目

删除项目

批量插入新项目

一对多集合

多对多集合

一对多集合

多对多集合

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

SqlAlchemy 2.0 中文文档（十三）（4）

写入仅集合 - API 文档

创建和持久化新的只写集合

向现有集合添加新项目

查询项目

删除项目

批量插入新项目

一对多集合

多对多集合

一对多集合

多对多集合

热门文章

最新文章

相关电子书