SQLAlchemy 的會(huì)話緩存（Session Cache）是 ORM 框架的核心特性之一，對(duì)于理解和高效使用 SQLAlchemy 至關(guān)重要。這個(gè)緩存機(jī)制主要作用在會(huì)話（Session）層面，提供了對(duì)數(shù)據(jù)庫(kù)交互的中間緩存層。以下是對(duì) SQLAlchemy 會(huì)話緩存的詳細(xì)解釋?zhuān)?/p>

什么是會(huì)話緩存？

1. 一級(jí)緩存：會(huì)話緩存也被稱(chēng)為一級(jí)緩存。它自動(dòng)存儲(chǔ)在一個(gè) SQLAlchemy Session 生命周期內(nèi)加載的所有 ORM 對(duì)象。這意味著在會(huì)話期間，對(duì)同一個(gè)數(shù)據(jù)庫(kù)實(shí)體的重復(fù)查詢不會(huì)導(dǎo)致多次數(shù)據(jù)庫(kù)請(qǐng)求。

2. 自動(dòng)化的工作流：當(dāng)你通過(guò)一個(gè)會(huì)話查詢數(shù)據(jù)庫(kù)時(shí)，SQLAlchemy 首先檢查這個(gè)對(duì)象是否已經(jīng)在會(huì)話緩存中。如果是，它會(huì)直接從緩存中返回對(duì)象，而不是從數(shù)據(jù)庫(kù)重新加載。

3. 對(duì)象唯一性：在一個(gè)會(huì)話中，對(duì)于具有相同主鍵的實(shí)體，會(huì)話緩存確保只有一個(gè)唯一的對(duì)象實(shí)例。這有助于維護(hù)數(shù)據(jù)的一致性。

會(huì)話緩存的工作原理

當(dāng)您使用 Session 對(duì)象查詢數(shù)據(jù)庫(kù)時(shí)，例如：

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

• 如果 my_object 是首次被請(qǐng)求，它會(huì)被加載并存儲(chǔ)在會(huì)話緩存中。

• 如果稍后在同一會(huì)話中再次查詢相同的 MyModel 實(shí)例，SQLAlchemy 會(huì)直接從會(huì)話緩存中返回這個(gè)對(duì)象，而不是執(zhí)行新的數(shù)據(jù)庫(kù)查詢。

會(huì)話緩存的好處

1. 減少數(shù)據(jù)庫(kù)查詢：通過(guò)減少對(duì)數(shù)據(jù)庫(kù)的重復(fù)查詢，提高了應(yīng)用性能。

2. 數(shù)據(jù)一致性：在會(huì)話期間，對(duì)于同一對(duì)象的更改在整個(gè)會(huì)話中是一致的，避免了可能的數(shù)據(jù)不一致問(wèn)題。

3. 事務(wù)支持：會(huì)話緩存支持事務(wù)操作。當(dāng)一個(gè)事務(wù)被回滾時(shí)，所有的會(huì)話緩存也會(huì)被回滾到事務(wù)開(kāi)始之前的狀態(tài)。

管理會(huì)話緩存

• 清空緩存：可以通過(guò) session.expire_all() 清空會(huì)話緩存，這會(huì)使所有已加載的對(duì)象變?yōu)椤斑^(guò)期”狀態(tài)，下次訪問(wèn)這些對(duì)象的任何屬性時(shí)，SQLAlchemy 會(huì)從數(shù)據(jù)庫(kù)重新加載它們。

• 手動(dòng)刷新：session.flush() 會(huì)將會(huì)話中的更改（如新對(duì)象或修改的對(duì)象）同步到數(shù)據(jù)庫(kù)，但不會(huì)提交事務(wù)。這不會(huì)影響會(huì)話緩存中已有的對(duì)象。

注意事項(xiàng)

• 長(zhǎng)期會(huì)話問(wèn)題：在長(zhǎng)期運(yùn)行的會(huì)話中，會(huì)話緩存可能會(huì)導(dǎo)致內(nèi)存占用增加，特別是在處理大量數(shù)據(jù)時(shí)。

• 數(shù)據(jù)過(guò)時(shí)問(wèn)題：如果數(shù)據(jù)庫(kù)中的數(shù)據(jù)在會(huì)話外被修改，會(huì)話緩存中的數(shù)據(jù)可能會(huì)過(guò)時(shí)。這種情況下，需要使用 expire、refresh 或 expire_all 方法來(lái)更新緩存數(shù)據(jù)。

代碼展示

print("=====================================會(huì)話緩存==================================================")

# 第一次查詢，并加載用戶的所有關(guān)聯(lián)部門(mén)項(xiàng)
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"用戶編號(hào)：{user1.id} 用戶姓名：{user1.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user1.depts]}")

# 第二次即使沒(méi)有加載用戶關(guān)聯(lián)的部門(mén)，同樣可以訪問(wèn)，因?yàn)檫@里會(huì)默認(rèn)從會(huì)話緩存中獲取
sql2 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset2 = await self.db.scalars(sql2)
user2 = queryset2.first()
print(f"用戶編號(hào)：{user2.id} 用戶姓名：{user2.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user2.depts]}")

# 使當(dāng)前會(huì)話（Session）中所有已加載的對(duì)象過(guò)期，確保您獲取的是數(shù)據(jù)庫(kù)中的最新數(shù)據(jù)。
self.db.expire_all()

print("===================查詢出來(lái)，即使沒(méi)有通過(guò).訪問(wèn)屬性，同樣會(huì)產(chǎn)生緩存=====================")

# 第一次查詢，并加載用戶的所有關(guān)聯(lián)部門(mén)項(xiàng)，但是不訪問(wèn)用戶的屬性
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"沒(méi)有訪問(wèn)屬性，也會(huì)產(chǎn)生緩存")

# 第二次即使沒(méi)有加載用戶關(guān)聯(lián)的部門(mén)，同樣可以訪問(wèn)，因?yàn)檫@里會(huì)默認(rèn)從會(huì)話緩存中獲取
sql4 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset4 = await self.db.scalars(sql4)
user4 = queryset4.first()
print(f"用戶編號(hào)：{user4.id} 用戶姓名：{user4.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user4.depts]}")

# 使當(dāng)前會(huì)話（Session）中所有已加載的對(duì)象過(guò)期，確保您獲取的是數(shù)據(jù)庫(kù)中的最新數(shù)據(jù)。
self.db.expire_all()

print("=====================================數(shù)據(jù)列表會(huì)話緩存==================================================")

# 第一次查詢出所有用戶，并加載用戶的所有關(guān)聯(lián)部門(mén)項(xiàng)
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"用戶編號(hào)：{data.id} 用戶姓名：{data.name} 關(guān)聯(lián)部門(mén) {[i.name for i in data.depts]}")

# 第二次即使沒(méi)有加載用戶關(guān)聯(lián)的部門(mén)，同樣可以訪問(wèn)，因?yàn)檫@里會(huì)默認(rèn)從會(huì)話緩存中獲取
sql6 = select(models.VadminUser)
queryset6 = await self.db.scalars(sql6)
datas6 = queryset6.unique().all()
for data in datas6:
    print(f"用戶編號(hào)：{data.id} 用戶姓名：{data.name} 關(guān)聯(lián)部門(mén) {[i.name for i in data.depts]}")

# 使當(dāng)前會(huì)話（Session）中所有已加載的對(duì)象過(guò)期，確保您獲取的是數(shù)據(jù)庫(kù)中的最新數(shù)據(jù)。
self.db.expire_all()

print("===================expire 單個(gè)對(duì)象過(guò)期=====================")

# 第一次查詢，并加載用戶的所有關(guān)聯(lián)部門(mén)項(xiàng)
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"用戶編號(hào)：{user7.id} 用戶姓名：{user7.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user7.depts]}")

# 使當(dāng)前會(huì)話（Session）中的 user7 對(duì)象過(guò)期，再次訪問(wèn)就會(huì)重新查詢數(shù)據(jù)庫(kù)數(shù)據(jù)
self.db.expire(user7)

# 第二次查詢會(huì)發(fā)現(xiàn)會(huì)話中沒(méi)有該對(duì)象的緩存，會(huì)重新在數(shù)據(jù)庫(kù)中查詢
sql8 = select(models.VadminUser).where(models.VadminUser.id == 1)
queryset8 = await self.db.scalars(sql8)
user8 = queryset8.first()
try:
    print(f"用戶編號(hào)：{user8.id} 用戶姓名：{user8.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user8.depts]}")
except StatementError:
    print("訪問(wèn)部門(mén)報(bào)錯(cuò)了?。。。?！")

# 使當(dāng)前會(huì)話（Session）中所有已加載的對(duì)象過(guò)期，確保您獲取的是數(shù)據(jù)庫(kù)中的最新數(shù)據(jù)。
self.db.expire_all()

print("=========expire 單個(gè)對(duì)象過(guò)期后，重新訪問(wèn)之前對(duì)象的屬性也會(huì)重新查詢數(shù)據(jù)庫(kù)，但是不會(huì)重新加載關(guān)系===========")

# 第一次查詢，并加載用戶的所有關(guān)聯(lián)部門(mén)項(xiàng)
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"用戶編號(hào)：{user9.id} 用戶姓名：{user9.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user9.depts]}")

# 使當(dāng)前會(huì)話（Session）中的 user9 對(duì)象過(guò)期，再次訪問(wèn)就會(huì)重新查詢數(shù)據(jù)庫(kù)數(shù)據(jù)
self.db.expire(user9)

# 第二次查詢會(huì)發(fā)現(xiàn)會(huì)話中沒(méi)有該對(duì)象的緩存，會(huì)重新在數(shù)據(jù)庫(kù)中查詢，但是不會(huì)重新加載關(guān)系
try:
    print(f"用戶編號(hào)：{user9.id} 用戶姓名：{user9.name} 關(guān)聯(lián)部門(mén) {[i.name for i in user9.depts]}")
except StatementError:
    print("訪問(wèn)部門(mén)報(bào)錯(cuò)了?。。。。?)

print("=====================================結(jié)束==================================================")

總結(jié)

會(huì)話緩存是 SQLAlchemy 中一個(gè)強(qiáng)大的特性，它提高了應(yīng)用性能并支持復(fù)雜的事務(wù)管理。然而，合理地管理會(huì)話和緩存是確保應(yīng)用穩(wěn)定性和數(shù)據(jù)一致性的關(guān)鍵。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-786450.html

到了這里，關(guān)于SQLAlchemy 中的會(huì)話（Session）緩存詳解的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！