在 SQLAlchemy 中，expire_all() 方法是 Session 类的一个重要方法，用于使会话（Session）中所有当前加载的对象过期。这意味着下次访问任何已加载对象的属性时，SQLAlchemy 将从数据库中重新加载这些对象的最新状态。这是一个强制刷新加载对象状态的方法，确保您获取的是数据库中的最新数据。
功能
当您调用 session.expire_all() 时，会发生以下事情：
-
标记所有对象为过期：在该会话中加载的所有对象都被标记为“过期”。这意味着它们的当前状态将被丢弃，而不是从会话的缓存中提取。
-
延迟加载：在下次访问这些对象的任何属性时，SQLAlchemy 会自动从数据库中重新加载它们的最新状态。这是一个延迟操作，只有在实际访问属性时才会触发。
-
不影响未修改的对象：即使对象自加载后未被修改，它们的状态也会被清除并在下次访问时重新加载。
使用场景
您可能会在以下情况下使用 expire_all()：
-
保证数据一致性：在长时间运行的会话中，为了确保获取到数据库中的最新数据，可以使用 expire_all() 方法。
-
避免脏读：在有可能发生并发修改的场景中，使用 expire_all() 可以防止读取到过时的数据。
-
手动刷新状态：在某些复杂的交互逻辑中，您可能需要手动刷新对象状态，以确保应用逻辑的正确性。文章来源：https://www.toymoban.com/news/detail-793946.html
示例
from sqlalchemy.orm import sessionmaker
from my_model import MyModel # 假设您有一个模型类 MyModel
from sqlalchemy import create_engine
# 创建会话
engine = create_engine("sqlite:///mydatabase.db")
Session = sessionmaker(bind=engine)
session = Session()
# 查询并使用对象
my_object = session.query(MyModel).first()
print(my_object.some_attribute)
# 确保从数据库重新加载对象
session.expire_all()
# 当再次访问属性时，将从数据库重新加载它的最新状态
print(my_object.some_attribute)
SQLAlchemy 2.0 代码展示
print("=====================================会话缓存==================================================")
# 第一次查询，并加载用户的所有关联部门项
sql1 = select(models.VadminUser).where(models.VadminUser.id == 1).options(joinedload(models.VadminUser.depts))
queryset1 = await self.db.scalars(sql1)
user1 = queryset1.unique().first()
print(f"用户编号：{user1.id} 用户姓名：{user1.name} 关联部门 {[i.name for i in user1.depts]}")
# 第二次即使没有加载用户关联的部门，同样可以访问，因为这里会默认从会话缓存中获取
sql2 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset2 = await self.db.scalars(sql2)
user2 = queryset2.first()
print(f"用户编号：{user2.id} 用户姓名：{user2.name} 关联部门 {[i.name for i in user2.depts]}")
# 使当前会话（Session）中所有已加载的对象过期，确保您获取的是数据库中的最新数据。
self.db.expire_all()
print("===================查询出来，即使没有通过.访问属性，同样会产生缓存=====================")
# 第一次查询，并加载用户的所有关联部门项，但是不访问用户的属性
sql3 = select(models.VadminUser).where(models.VadminUser.id == 1).options(joinedload(models.VadminUser.depts))
queryset3 = await self.db.scalars(sql3)
user3 = queryset3.unique().first()
print(f"没有访问属性，也会产生缓存")
# 第二次即使没有加载用户关联的部门，同样可以访问，因为这里会默认从会话缓存中获取
sql4 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset4 = await self.db.scalars(sql4)
user4 = queryset4.first()
print(f"用户编号：{user4.id} 用户姓名：{user4.name} 关联部门 {[i.name for i in user4.depts]}")
# 使当前会话（Session）中所有已加载的对象过期，确保您获取的是数据库中的最新数据。
self.db.expire_all()
print("=====================================数据列表会话缓存==================================================")
# 第一次查询出所有用户，并加载用户的所有关联部门项
sql5 = select(models.VadminUser).options(joinedload(models.VadminUser.depts))
queryset5 = await self.db.scalars(sql5)
datas5 = queryset5.unique().all()
for data in datas5:
print(f"用户编号：{data.id} 用户姓名：{data.name} 关联部门 {[i.name for i in data.depts]}")
# 第二次即使没有加载用户关联的部门，同样可以访问，因为这里会默认从会话缓存中获取
sql6 = select(models.VadminUser)
queryset6 = await self.db.scalars(sql6)
datas6 = queryset6.unique().all()
for data in datas6:
print(f"用户编号：{data.id} 用户姓名：{data.name} 关联部门 {[i.name for i in data.depts]}")
# 使当前会话（Session）中所有已加载的对象过期，确保您获取的是数据库中的最新数据。
self.db.expire_all()
print("===================expire 单个对象过期=====================")
# 第一次查询，并加载用户的所有关联部门项
sql7 = select(models.VadminUser).where(models.VadminUser.id == 1).options(joinedload(models.VadminUser.depts))
queryset7 = await self.db.scalars(sql7)
user7 = queryset7.unique().first()
print(f"用户编号：{user7.id} 用户姓名：{user7.name} 关联部门 {[i.name for i in user7.depts]}")
# 使当前会话（Session）中的 user7 对象过期，再次访问就会重新查询数据库数据
self.db.expire(user7)
# 第二次查询会发现会话中没有该对象的缓存，会重新在数据库中查询
sql8 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset8 = await self.db.scalars(sql8)
user8 = queryset8.first()
try:
print(f"用户编号：{user8.id} 用户姓名：{user8.name} 关联部门 {[i.name for i in user8.depts]}")
except StatementError:
print("访问部门报错了！！！！！")
# 使当前会话（Session）中所有已加载的对象过期，确保您获取的是数据库中的最新数据。
self.db.expire_all()
print("=========expire 单个对象过期后，重新访问之前对象的属性也会重新查询数据库，但是不会重新加载关系===========")
# 第一次查询，并加载用户的所有关联部门项
sql9 = select(models.VadminUser).where(models.VadminUser.id == 1).options(joinedload(models.VadminUser.depts))
queryset9 = await self.db.scalars(sql9)
user9 = queryset9.unique().first()
print(f"用户编号：{user9.id} 用户姓名：{user9.name} 关联部门 {[i.name for i in user9.depts]}")
# 使当前会话（Session）中的 user9 对象过期，再次访问就会重新查询数据库数据
self.db.expire(user9)
# 第二次查询会发现会话中没有该对象的缓存，会重新在数据库中查询，但是不会重新加载关系
try:
print(f"用户编号：{user9.id} 用户姓名：{user9.name} 关联部门 {[i.name for i in user9.depts]}")
except StatementError:
print("访问部门报错了！！！！！")
print("=====================================结束==================================================")
在这个示例中，使用 expire_all() 后，当再次访问 my_object.some_attribute 时，SQLAlchemy 将从数据库中重新加载 MyModel 对象的最新状态。文章来源地址https://www.toymoban.com/news/detail-793946.html
到了这里，关于sqlalchemy expire_all 方法详解，强制刷新会话缓存的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
