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

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

状态管理

原文：docs.sqlalchemy.org/en/20/orm/session_state_management.html

对象状态简介

知道对象在会话中可能具有的状态是有帮助的：

• Transient - 一个不在会话中的实例，也没有保存到数据库；即它没有数据库标识。这样的对象与 ORM 的唯一关系是其类与一个Mapper相关联。
  
• Pending - 当您Session.add()一个瞬态实例时，它变为待定状态。它实际上还没有被刷新到数据库中，但在下一次刷新时会被刷新到数据库中。
  
• Persistent - 存在于会话中并在数据库中具有记录的实例。您可以通过刷新使待定实例变为持久实例，或者通过查询数据库获取现有实例（或将其他会话中的持久实例移动到您的本地会话中）来获取持久实例。
  
• Deleted - 在刷新中已被删除的实例，但事务尚未完成。处于此状态的对象基本上处于“待定”状态的相反状态；当会话的事务提交时，对象将移动到分离状态。或者，当会话的事务回滚时，删除的对象将返回到持久状态。
  
• Detached - 一个对应于数据库中的记录，但目前不在任何会话中的实例。分离的对象将包含一个数据库标识标记，但是由于它没有与会话关联，因此无法确定此数据库标识是否实际存在于目标数据库中。分离的对象通常可以安全使用，但它们无法加载未加载的属性或先前标记为“过期”的属性。
  

深入了解所有可能的状态转换，请参阅对象生命周期事件部分，其中描述了每个转换以及如何以编程方式跟踪每个转换。

获取对象的当前状态

您可以随时使用inspect()函数在任何映射对象上查看实际状态；此函数将返回管理对象的内部 ORM 状态的相应InstanceState对象。InstanceState提供了其他访问器，包括指示对象持久状态的布尔属性，包括：

• InstanceState.transient
  
• InstanceState.pending
  
• InstanceState.persistent
  
• InstanceState.deleted
  
• InstanceState.detached
  

例如：

>>> from sqlalchemy import inspect
>>> insp = inspect(my_object)
>>> insp.persistent
True

另请参阅

映射实例的检查 - InstanceState 的更多示例 ## 会话属性

Session 本身在某种程度上就像一个集合。可以使用迭代器接口访问所有已存在的项目：

for obj in session:
    print(obj)

并且可以使用常规的“包含”语义来测试存在性：

if obj in session:
    print("Object is present")

会话还跟踪所有新创建的（即待处理的）对象，自上次加载或保存以来发生了更改的所有对象（即“脏对象”），以及标记为已删除的所有对象：

# pending objects recently added to the Session
session.new
# persistent objects which currently have changes detected
# (this collection is now created on the fly each time the property is called)
session.dirty
# persistent objects that have been marked as deleted via session.delete(obj)
session.deleted
# dictionary of all persistent objects, keyed on their
# identity key
session.identity_map

（文档：Session.new, Session.dirty, Session.deleted, Session.identity_map). ## 会话引用行为

会话内的对象是弱引用的。这意味着当它们在外部应用程序中取消引用时，它们也从Session 中消失，并且受 Python 解释器的垃圾收集影响。这种情况的例外包括待处理的对象、标记为已删除的对象或具有待处理更改的持久对象。在完全刷新后，这些集合都为空，并且所有对象再次成为弱引用。

为了使Session中的对象保持强引用，通常只需要一个简单的方法。外部管理强引用行为的示例包括将对象加载到以其主键为键的本地字典中，或者在它们需要保持引用的时间段内加载到列表或集合中。如果需要，这些集合可以与 Session 关联，方法是将它们放入 Session.info 字典中。

也可以采用基于事件的方法。一个简单的方法可以为所有对象在持久状态下保持“强引用”行为，具体如下：

from sqlalchemy import event
def strong_reference_session(session):
    @event.listens_for(session, "pending_to_persistent")
    @event.listens_for(session, "deleted_to_persistent")
    @event.listens_for(session, "detached_to_persistent")
    @event.listens_for(session, "loaded_as_persistent")
    def strong_ref_object(sess, instance):
        if "refs" not in sess.info:
            sess.info["refs"] = refs = set()
        else:
            refs = sess.info["refs"]
        refs.add(instance)
    @event.listens_for(session, "persistent_to_detached")
    @event.listens_for(session, "persistent_to_deleted")
    @event.listens_for(session, "persistent_to_transient")
    def deref_object(sess, instance):
        sess.info["refs"].discard(instance)

在上面的示例中，我们拦截了 SessionEvents.pending_to_persistent()、SessionEvents.detached_to_persistent()、SessionEvents.deleted_to_persistent() 和 SessionEvents.loaded_as_persistent() 事件钩子，以便拦截对象在进入持久状态时的行为，并在对象离开持久状态时拦截 SessionEvents.persistent_to_detached() 和 SessionEvents.persistent_to_deleted() 事件钩子。

上述函数可用于任何 Session，以在每个Session基础上提供强引用行为：

from sqlalchemy.orm import Session
my_session = Session()
strong_reference_session(my_session)

对于任何 sessionmaker，也可能被调用：

from sqlalchemy.orm import sessionmaker
maker = sessionmaker()
strong_reference_session(maker)
```## 合并
`Session.merge()` 将状态从外部对象传输到会话中的新实例或已存在的实例。它还将传入的数据与数据库状态进行对比，生成一个历史流，该流将被应用于下一次刷新，或者可以被设置为生成简单的状态“传输”，而不生成变更历史或访问数据库。使用方法如下：
```py
merged_object = session.merge(existing_object)

给定一个实例时，它遵循以下步骤：

• 它检查实例的主键。如果存在，则尝试在本地标识映射中定位该实例。如果 load=True 标志保持默认设置，则还会检查数据库是否存在此主键，如果在本地找不到，则检查数据库是否存在此主键。
  
• 如果给定实例没有主键，或者给定的主键找不到实例，则创建一个新实例。
  
• 然后将给定实例的状态复制到定位的/新创建的实例上。对于源实例上存在的属性值，该值将转移到目标实例上。对于源实例上不存在的属性值，目标实例上的相应属性将从内存中过期，这会丢弃目标实例的该属性的任何本地存在值，但不会对该属性的数据库持久化值进行直接修改。
  如果 load=True 标志保持默认设置，此复制过程会发出事件，并且将为源对象上的每个属性加载目标对象的未加载集合，以便可以根据数据库中存在的内容来协调传入状态。如果传递 load 为 False，则传入的数据将直接“标记”，而不产生任何历史记录。
  
• 操作会根据 merge 级联（请参阅级联）传播到相关对象和集合。
  
• 返回新实例。
  

使用Session.merge()，给定的“源”实例不会被修改，也不会与目标Session关联，并且仍然可以与任意数量的其他Session对象合并。Session.merge()对于获取任何类型的对象结构的状态而无需考虑其来源或当前会话关联，并将其状态复制到新会话中非常有用。以下是一些示例：

• 从文件读取对象结构并希望将其保存到数据库的应用程序可能会解析文件，构建结构，然后使用Session.merge()将其保存到数据库中，确保文件中的数据用于构造结构的每个元素的主键。稍后，当文件发生更改时，可以重新运行相同的过程，生成稍微不同的对象结构，然后可以再次进行merge，并且Session将自动更新数据库以反映这些更改，通过主键从数据库加载每个对象，然后使用新状态更新其状态。
  
• 一个应用程序将对象存储在一个内存缓存中，由许多Session对象同时共享。每次从缓存中检索对象时，都会使用Session.merge()创建它的本地副本，以便在每个请求它的Session中。缓存的对象保持分离状态；只有它的状态被移动到本地于各个Session对象的副本中。
  在缓存用例中，通常使用load=False标志来消除对象状态与数据库之间的开销。还有一个“批量”版本的Session.merge()称为Query.merge_result()，它被设计用于与缓存扩展的Query对象一起使用 - 请参阅 Dogpile Caching 部分。
  
• 一个应用程序想要将一系列对象的状态转移到由工作线程或其他并发系统维护的Session中。Session.merge()将每个要放入新Session中的对象复制一份。操作结束时，父线程/进程保留了其开始的对象，而线程/工作程序可以继续使用这些对象的本地副本。
  在“线程/进程之间传输”用例中，应用程序可能希望同时使用load=False标志，以避免开销和冗余的 SQL 查询，因为数据正在传输。
  

合并提示

Session.merge()是一个非常有用的方法，适用于许多目的。然而，它处理的是瞬态/分离对象和持久化对象之间复杂的边界，以及状态的自动转移。这里可能出现的各种各样的场景通常需要对对象状态更加谨慎的处理。合并的常见问题通常涉及到传递给Session.merge()的对象的一些意外状态。

让我们以用户和地址对象的典型示例为例：

class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(50), nullable=False)
    addresses = relationship("Address", backref="user")
class Address(Base):
    __tablename__ = "address"
    id = mapped_column(Integer, primary_key=True)
    email_address = mapped_column(String(50), nullable=False)
    user_id = mapped_column(Integer, ForeignKey("user.id"), nullable=False)

假设一个具有一个地址的User对象，已经持久化：

>>> u1 = User(name="ed", addresses=[Address(email_address="ed@ed.com")])
>>> session.add(u1)
>>> session.commit()

现在我们创建了一个在会话之外的对象a1，我们希望将其合并到现有的Address上：

>>> existing_a1 = u1.addresses[0]
>>> a1 = Address(id=existing_a1.id)

如果我们这样说会有一个意外的情况：

>>> a1.user = u1
>>> a1 = session.merge(a1)
>>> session.commit()
sqlalchemy.orm.exc.FlushError: New instance <Address at 0x1298f50>
with identity key (<class '__main__.Address'>, (1,)) conflicts with
persistent instance <Address at 0x12a25d0>

为什么会这样？我们没有正确处理级联。将a1.user分配给持久对象级联到User.addresses的反向引用，并使我们的a1对象挂起，就好像我们已经添加了它一样。现在我们的会话中有两个Address对象：

>>> a1 = Address()
>>> a1.user = u1
>>> a1 in session
True
>>> existing_a1 in session
True
>>> a1 is existing_a1
False

上面，我们的a1已经在会话中挂起。随后的Session.merge()操作实际上什么都不做。级联可以通过relationship.cascade选项在relationship()上配置，尽管在这种情况下，这意味着从User.addresses关系中删除了save-update级联 - 而且通常，那种行为非常方便。这里的解决方案通常是不将a1.user分配给已经存在于目标会话中的对象。

relationship()的cascade_backrefs=False选项也将阻止通过a1.user = u1分配将Address添加到会话中。

更多关于级联操作的细节请参阅级联。

另一个意外状态的例子：

>>> a1 = Address(id=existing_a1.id, user_id=u1.id)
>>> a1.user = None
>>> a1 = session.merge(a1)
>>> session.commit()
sqlalchemy.exc.IntegrityError: (IntegrityError) address.user_id
may not be NULL

上面，将user的分配优先于user_id的外键分配，最终导致user_id应用了None，导致失败。

大多数Session.merge()问题可以通过首先检查 - 对象是否过早地在会话中？

>>> a1 = Address(id=existing_a1, user_id=user.id)
>>> assert a1 not in session
>>> a1 = session.merge(a1)

或者对象上有我们不想要的状态吗？检查__dict__是一个快速检查的方法：

>>> a1 = Address(id=existing_a1, user_id=user.id)
>>> a1.user
>>> a1.__dict__
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1298d10>,
 'user_id': 1,
 'id': 1,
 'user': None}
>>> # we don't want user=None merged, remove it
>>> del a1.user
>>> a1 = session.merge(a1)
>>> # success
>>> session.commit()

清除

Expunge 将对象从会话中删除，将持久实例发送到脱机状态，将待处理实例发送到瞬态状态：

session.expunge(obj1)

要删除所有项目，请调用Session.expunge_all()（此方法以前称为clear()）。

刷新 / 过期

过期意味着数据库持久化数据存储在一系列对象属性中被清除，这样当下次访问这些属性时，将发出一个 SQL 查询，该查询将从数据库中刷新数据。

当我们谈论数据的过期时，我们通常是指处于持久状态的对象。例如，如果我们像这样加载一个对象：

user = session.scalars(select(User).filter_by(name="user1").limit(1)).first()

上述的User对象是持久的，并且具有一系列存在的属性；如果我们查看它的__dict__，我们会看到已加载的状态：

>>> user.__dict__
{
 'id': 1, 'name': u'user1',
 '_sa_instance_state': <...>,
}

其中id和name是数据库中的列。_sa_instance_state是 SQLAlchemy 内部使用的非数据库持久化值（它引用了实例的InstanceState。虽然与本节直接相关，但如果我们想要获取它，我们应该使用inspect()函数来访问它）。

此时，我们User对象中的状态与加载的数据库行的状态相匹配。但是在使用诸如Session.expire()之类的方法使对象过期后，我们会看到状态被删除：

>>> session.expire(user)
>>> user.__dict__
{'_sa_instance_state': <...>}

我们看到，虽然内部的“状态”仍然存在，但与id和name列对应的值已经消失。如果我们要访问其中一列并观察 SQL，我们会看到这样的情况：

>>> print(user.name)
SELECT  user.id  AS  user_id,  user.name  AS  user_name
FROM  user
WHERE  user.id  =  ?
(1,)
user1

上面，在访问已过期的属性user.name时，ORM 启动了一个惰性加载以从数据库中检索最新状态，通过向这个用户引用的用户行发出一个 SELECT。之后，__dict__再次被填充：

>>> user.__dict__
{
 'id': 1, 'name': u'user1',
 '_sa_instance_state': <...>,
}

注意

虽然我们正在查看__dict__的内容，以便了解 SQLAlchemy 对对象属性的处理，但我们不应直接修改__dict__的内容，至少不应修改 SQLAlchemy ORM 正在维护的属性（SQLA 领域之外的其他属性没问题）。这是因为 SQLAlchemy 使用描述符来跟踪我们对对象所做的更改，当我们直接修改__dict__时，ORM 将无法跟踪到我们做出的更改。

Session.expire()和Session.refresh()的另一个关键行为是，对象上的所有未刷新的更改都将被丢弃。也就是说，如果我们要修改User上的属性：

>>> user.name = "user2"

但是当我们在调用Session.expire()之前没有调用Session.flush()时，我们挂起的值'user2'将被丢弃：

>>> session.expire(user)
>>> user.name
'user1'

Session.expire()方法可用于将实例的所有 ORM 映射属性标记为“过期”：

# expire all ORM-mapped attributes on obj1
session.expire(obj1)

它还可以传递一个字符串属性名称列表，指定要标记为过期的特定属性：

# expire only attributes obj1.attr1, obj1.attr2
session.expire(obj1, ["attr1", "attr2"])

Session.expire_all()方法允许我们一次性对Session中包含的所有对象调用Session.expire()：

session.expire_all()

Session.refresh()方法具有类似的接口，但是不是使过期，而是立即发出对象行的 SELECT：

# reload all attributes on obj1
session.refresh(obj1)

Session.refresh()还接受一个字符串属性名称的列表，但与Session.expire()不同，它期望至少一个名称是列映射属性的名称：

# reload obj1.attr1, obj1.attr2
session.refresh(obj1, ["attr1", "attr2"])

提示

通常更灵活的刷新替代方法是使用 ORM 的填充现有内容功能，适用于使用select()进行 2.0 风格查询以及在 1.x 风格查询中的Query.populate_existing()方法。使用此执行选项，语句结果集中返回的所有 ORM 对象都将使用来自数据库的数据进行刷新：

stmt = (
    select(User)
    .execution_options(populate_existing=True)
    .where((User.name.in_(["a", "b", "c"])))
)
for user in session.execute(stmt).scalars():
    print(user)  # will be refreshed for those columns that came back from the query

查看填充现有内容以获取更多详细信息。

实际加载内容

当标记为Session.expire()或使用Session.refresh()加载的对象时，所发出的 SELECT 语句因多种因素而异，包括：

• 仅从列映射属性加载过期属性。虽然可以将任何类型的属性标记为过期，包括relationship() - 映射属性，但访问过期的relationship()属性将仅为该属性发出加载，使用标准的基于关系的惰性加载。即使过期，基于列的属性也不会作为此操作的一部分加载，而是在访问任何基于列的属性时加载。
  
• 通过 relationship() 映射的属性不会在访问过期的基于列的属性时加载。
  
• 关于关系，Session.refresh() 在属性不是列映射的情况下比 Session.expire() 更为严格。调用 Session.refresh() 并传递一个只包括关系映射属性的名称列表将会引发错误。无论如何，非急切加载的 relationship() 属性都不会包含在任何刷新操作中。
  
• 通过 relationship.lazy 参数配置为“急切加载”的 relationship() 属性将在 Session.refresh() 的情况下加载，如果未指定任何属性名称，或者如果它们的名称包含在要刷新的属性列表中。
  
• 配置为 deferred() 的属性通常不会在过期属性加载期间或刷新期间加载。当直接访问未加载的属性或者作为延迟属性组的一部分访问该组中的未加载属性时，配置为 deferred() 的未加载属性将自行加载。
  
• 对于在访问时加载的过期属性，连接继承表映射将发出一个通常只包含那些存在未加载属性的表的 SELECT。在这里的操作足够复杂，以仅加载父表或子表，例如，如果最初过期的列的子集仅包含其中一个表或另一个表。
  
• 当在连接继承表映射上使用 Session.refresh() 时，所发出的 SELECT 与在目标对象的类上使用 Session.query() 时的类似。这通常是映射的一部分设置的所有表。
  

何时过期或刷新

Session 在会话引用的事务结束时自动使用过期功能。这意味着，每当调用 Session.commit() 或 Session.rollback()，会话中的所有对象都会过期，使用与 Session.expire_all() 方法相当的功能。其原因在于事务的结束是一个标志性的点，在此点上不再有可用于了解数据库当前状态的上下文，因为任意数量的其他事务可能正在影响它。只有当新事务开始时，我们才能再次访问数据库的当前状态，在此时可能已经发生了任意数量的更改。

当希望强制对象重新从数据库加载其数据时，应使用 Session.expire() 和 Session.refresh() 方法，当已知数据的当前状态可能过时时。这样做的原因可能包括：

• 一些 SQL 在 ORM 对象处理范围之外的事务中被发出，比如如果使用 Session.execute() 方法发出了 Table.update() 构造；
  
• 如果应用程序试图获取已知在并发事务中已修改的数据，并且已知正在生效的隔离规则允许该数据可见。
  

第二条警告很重要，即“也已知在生效的隔离规则下，这些数据可见。”这意味着不能假设在另一个数据库连接上发生的更新在本地已经可见；在许多情况下，它是不可见的。这就是为什么如果想要在正在进行的事务之间使用 Session.expire() 或 Session.refresh() 查看数据，就必须了解正在生效的隔离行为。

另见

Session.expire()

Session.expire_all()

Session.refresh()

填充现有 - 允许任何 ORM 查询在 SELECT 语句的结果中刷新对象，就像它们通常加载一样，刷新标识映射中所有匹配的对象。

隔离 - 隔离的词汇解释，其中包括指向维基百科的链接。

SQLAlchemy 会话深入解析 - 一个关于对象生命周期的深入讨论的视频 + 幻灯片，包括数据过期的角色。## 快速对象状态介绍

了解实例在会话中可能具有的状态是有帮助的：

• 瞬时 - 一个不在会话中并且没有保存到数据库的实例；即它没有数据库标识。这样的对象与 ORM 的唯一关系是其类与一个 Mapper 相关联。
  
• 待定 - 当你 Session.add() 一个瞬时实例时，它变为待定状态。它实际上还没有被刷新到数据库，但在下一次刷新时会被刷新到数据库。
  
• 持久 - 存在于会话中并且在数据库中有记录的实例。您可以通过刷新使待定实例变为持久实例，或通过查询数据库获取现有实例（或将其他会话中的持久实例移动到您的本地会话）来获得持久实例。
  
• 已删除 - 在刷新中已删除的实例，但事务尚未完成。处于这种状态的对象基本上与“待定”状态相反；当会话的事务提交时，对象将移至分离状态。另外，当会话的事务回滚时，已删除的对象将回到持久状态。
  
• 分离 -  一个实例，它对应于或以前对应于数据库中的记录，但当前不在任何会话中。分离的对象将包含一个数据库标识标记，但由于它没有关联到会话，因此不知道此数据库标识实际上是否存在于目标数据库中。分离的对象通常是安全的使用，除了它们无法加载未加载的属性或以前标记为“过期”的属性。
  

深入研究所有可能的状态转换，请参阅 对象生命周期事件 部分，该部分描述了每个转换以及如何以编程方式跟踪每个转换。

获取对象的当前状态

任何映射对象的实际状态都可以随时使用映射实例上的 inspect() 函数查看；此函数将返回管理对象的内部 ORM 状态的相应 InstanceState 对象。InstanceState 提供了其他访问器，其中包括指示对象持久性状态的布尔属性，包括：

• InstanceState.transient
  
• InstanceState.pending
  
• InstanceState.persistent
  
• InstanceState.deleted
  
• InstanceState.detached
  

例如：

>>> from sqlalchemy import inspect
>>> insp = inspect(my_object)
>>> insp.persistent
True

另请参阅

映射实例的检查 - 更多有关 InstanceState 的示例

获取对象的当前状态

任何映射对象的实际状态都可以随时使用映射实例上的 inspect() 函数查看；此函数将返回管理对象的内部 ORM 状态的相应 InstanceState 对象。InstanceState 提供了其他访问器，其中包括指示对象持久性状态的布尔属性，包括：

• InstanceState.transient
  
• InstanceState.pending
  
• InstanceState.persistent
  
• InstanceState.deleted
  
• InstanceState.detached
  

例如：

>>> from sqlalchemy import inspect
>>> insp = inspect(my_object)
>>> insp.persistent
True

另请参阅

映射实例的检查 - 更多有关 InstanceState 的示例

会话属性

Session 本身的行为有点像一个类似集合的集合。可以使用迭代器接口访问所有存在的项目：

for obj in session:
    print(obj)

并且可以使用常规的“包含”语义进行测试：

if obj in session:
    print("Object is present")

会话还会跟踪所有新创建的（即待处理的）对象，所有自上次加载或保存以来发生更改的对象（即“脏对象”），以及所有被标记为已删除的对象：

# pending objects recently added to the Session
session.new
# persistent objects which currently have changes detected
# (this collection is now created on the fly each time the property is called)
session.dirty
# persistent objects that have been marked as deleted via session.delete(obj)
session.deleted
# dictionary of all persistent objects, keyed on their
# identity key
session.identity_map

（文档：Session.new，Session.dirty，Session.deleted，Session.identity_map）。

会话引用行为

会话中的对象是弱引用的。这意味着当它们在外部应用程序中取消引用时，它们也会从Session中失去作用域，并且由 Python 解释器进行垃圾回收。这种情况的例外包括待处理对象、标记为已删除的对象或具有待处理更改的持久对象。在完全刷新后，这些集合都为空，并且所有对象再次是弱引用的。

使Session中的对象保持强引用通常只需要简单的方法。外部管理的强引用行为示例包括将对象加载到以其主键为键的本地字典中，或者在它们需要保持引用的时间段内加载到列表或集合中。如果需要，这些集合可以与Session关联，方法是将它们放入Session.info字典中。

还可以使用基于事件的方法。以下是一个提供了所有对象在持久化状态下保持“强引用”行为的简单方案：

from sqlalchemy import event
def strong_reference_session(session):
    @event.listens_for(session, "pending_to_persistent")
    @event.listens_for(session, "deleted_to_persistent")
    @event.listens_for(session, "detached_to_persistent")
    @event.listens_for(session, "loaded_as_persistent")
    def strong_ref_object(sess, instance):
        if "refs" not in sess.info:
            sess.info["refs"] = refs = set()
        else:
            refs = sess.info["refs"]
        refs.add(instance)
    @event.listens_for(session, "persistent_to_detached")
    @event.listens_for(session, "persistent_to_deleted")
    @event.listens_for(session, "persistent_to_transient")
    def deref_object(sess, instance):
        sess.info["refs"].discard(instance)

上述，我们拦截了SessionEvents.pending_to_persistent()，SessionEvents.detached_to_persistent()，SessionEvents.deleted_to_persistent()和SessionEvents.loaded_as_persistent()事件钩子，以拦截对象进入 persistent 状态转换时的情况，以及SessionEvents.persistent_to_detached()和SessionEvents.persistent_to_deleted()钩子以拦截对象离开持久状态时的情况。

上述函数可针对任何Session进行调用，以在每个Session上提供强引用行为：

from sqlalchemy.orm import Session
my_session = Session()
strong_reference_session(my_session)

也可以针对任何sessionmaker进行调用：

from sqlalchemy.orm import sessionmaker
maker = sessionmaker()
strong_reference_session(maker)

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