SQLAlchemy 中的会话（Session）缓存详解

SQLAlchemy 的会话缓存（Session Cache）是 ORM 框架的核心特性之一，对于理解和高效使用 SQLAlchemy 至关重要。这个缓存机制主要作用在会话（Session）层面，提供了对数据库交互的中间缓存层。以下是对 SQLAlchemy 会话缓存的详细解释：

什么是会话缓存？

1. 一级缓存：会话缓存也被称为一级缓存。它自动存储在一个 SQLAlchemy Session 生命周期内加载的所有 ORM 对象。这意味着在会话期间，对同一个数据库实体的重复查询不会导致多次数据库请求。

2. 自动化的工作流：当你通过一个会话查询数据库时，SQLAlchemy 首先检查这个对象是否已经在会话缓存中。如果是，它会直接从缓存中返回对象，而不是从数据库重新加载。

3. 对象唯一性：在一个会话中，对于具有相同主键的实体，会话缓存确保只有一个唯一的对象实例。这有助于维护数据的一致性。

会话缓存的工作原理

当您使用 Session 对象查询数据库时，例如：

my_object = session.query(MyModel).filter_by(id=1).first()

• 如果 my_object 是首次被请求，它会被加载并存储在会话缓存中。

• 如果稍后在同一会话中再次查询相同的 MyModel 实例，SQLAlchemy 会直接从会话缓存中返回这个对象，而不是执行新的数据库查询。

  会话缓存的好处

  1. 减少数据库查询：通过减少对数据库的重复查询，提高了应用性能。

  2. 数据一致性：在会话期间，对于同一对象的更改在整个会话中是一致的，避免了可能的数据不一致问题。

  3. 事务支持：会话缓存支持事务操作。当一个事务被回滚时，所有的会话缓存也会被回滚到事务开始之前的状态。

  管理会话缓存

  • 清空缓存：可以通过 session.expire_all() 清空会话缓存，这会使所有已加载的对象变为“过期”状态，下次访问这些对象的任何属性时，SQLAlchemy 会从数据库重新加载它们。

  • 手动刷新：session.flush() 会将会话中的更改（如新对象或修改的对象）同步到数据库，但不会提交事务。这不会影响会话缓存中已有的对象。

    注意事项

    • 长期会话问题：在长期运行的会话中，会话缓存可能会导致内存占用增加，特别是在处理大量数据时。

    • 数据过时问题：如果数据库中的数据在会话外被修改，会话缓存中的数据可能会过时。这种情况下，需要使用 expire、refresh 或 expire_all 方法来更新缓存数据。

      代码展示

      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("=====================================结束==================================================")
      

      总结

      会话缓存是 SQLAlchemy 中一个强大的特性，它提高了应用性能并支持复杂的事务管理。然而，合理地管理会话和缓存是确保应用稳定性和数据一致性的关键。