数据库、缓存与对象存储

更新 2026-10-05

正在读取学习记录…

工作流要保存的东西有不同形态。数据库适合记录任务标识、版本、状态和引用关系；对象存储适合保存较大的报告正文、图片和附件；缓存保存可以重新获取的副本，加快频繁读取。三者组合时，读者需要知道数据的权威来源，以及缓存失效后从哪里恢复。

例如报告列表可以从数据库查询每份报告的状态和正文对象键，打开报告时再加载正文。缓存让同一版本再次打开更快，更新报告则需要新的版本标识，避免把旧正文当成最新结果。

保存两版报告，再清空缓存重读

固定报告 report-42 第一版正文是“销售额100元”，第二版改为“销售额120元”。元数据存SQLite，正文以内容哈希命名保存在文件夹中。内容哈希是将正文计算成的固定指纹，相同正文得到相同名称，内容改变时名称也改变。这里用本地文件保存正文，缓存是进程内字典。

程序先读两次第一版，保存第二版，再读一次，最后清空缓存、关闭数据库并重新打开读取。每行打印版本、读取来源和正文，让三种存储职责直接可见。

元数据选择版本，再查缓存或正文，写入先正文后引用；缓存清空仍能恢复 — 图1：写入先正文后引用；缓存清空仍能恢复。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-storage" -Force
cd "F:\学习\l4-storage"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-storage 文件夹，把下面完整代码保存为 storage_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-storage"
ls
python -X utf8 storage_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-storage" -Force
cd "F:\学习\l4-storage"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""数据库存元数据，文件存正文，缓存按版本加速读取。"""
import hashlib
import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory


class Reports:
    def __init__(self, folder):
        self.folder = Path(folder)
        self.db = sqlite3.connect(self.folder / "metadata.db")
        self.db.execute("CREATE TABLE IF NOT EXISTS report (id TEXT PRIMARY KEY, version INTEGER, object_key TEXT)")
        self.db.commit()
        self.cache = {}

    def save(self, report_id, content):
        # 内容哈希作为对象键；先落正文，再提交元数据引用。
        key = hashlib.sha256(content.encode("utf-8")).hexdigest() + ".txt"
        path = self.folder / key
        path.write_text(content, encoding="utf-8")
        row = self.db.execute("SELECT version FROM report WHERE id=?", (report_id,)).fetchone()
        version = 1 if row is None else row[0] + 1
        with self.db:
            self.db.execute("INSERT OR REPLACE INTO report VALUES (?, ?, ?)",
                            (report_id, version, key))
        return version

    def read(self, report_id):
        row = self.db.execute("SELECT version, object_key FROM report WHERE id=?", (report_id,)).fetchone()
        if row is None:
            raise KeyError(report_id)
        version, key = row
        cache_key = (report_id, version)
        if cache_key in self.cache:
            return self.cache[cache_key], "cache", version
        content = (self.folder / key).read_text(encoding="utf-8")
        self.cache[cache_key] = content
        return content, "object", version

    def close(self):
        self.db.close()


def run_demo():
    with TemporaryDirectory(prefix="reports-") as folder:
        reports = Reports(folder)
        reports.save("report-42", "销售额100元")
        lines = []
        for label in ("首次读取", "再次读取"):
            text, source, version = reports.read("report-42")
            lines.append(f"{label}: v{version}, {source}, {text}")
        reports.save("report-42", "销售额120元")
        text, source, version = reports.read("report-42")
        lines.append(f"更新后读取: v{version}, {source}, {text}")
        reports.cache.clear()
        reports.close()
        reports = Reports(folder)
        text, source, version = reports.read("report-42")
        lines.append(f"重启并清空缓存: v{version}, {source}, {text}")
        reports.close()
        return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

首次读取: v1, object, 销售额100元
再次读取: v1, cache, 销售额100元
更新后读取: v2, object, 销售额120元
重启并清空缓存: v2, object, 销售额120元

缓存怎样识别报告已经更新

第一次读取来源为 object，因为正文来自文件；第二次为 cache，说明相同版本命中缓存。保存第二版后，数据库版本从1升到2，缓存键从 ("report-42", 1) 变成 ("report-42", 2)，于是再次读取正文对象。最后一行重启后仍有第二版内容，缓存丢失没有丢报告。

read 每次先读取数据库当前版本，随后查询该版本缓存。这需要一次元数据查询，却能在单工作者示例中避免旧正文命中。若同时把元数据也缓存，需明确失效规则，否则正文版本键再正确，也可能一直拿到旧版本号。

演示的数据库和正文放在系统配置的临时目录中，结束后自动清理。需要放在F盘时，在当前PowerShell运行 $env:TEMP = $env:TMP = "F:\学习\l4-storage"，再运行程序；这不会修改其他窗口或全局设置。正式报告应选择长期保存的目录和保留策略。

为什么先保存正文，再保存引用

save 先写对象文件，再提交数据库行。正文写入失败时，元数据不会引用一个尚不存在的对象。正文成功、数据库失败则会留下未引用的文件，可以定期检查并清理。数据库与远端对象服务通常无法用一个普通事务一起提交，需要围绕失败窗口安排写入顺序。

内容哈希把相同正文指向相同文件。示例不覆盖不同内容的同一对象键，适合不可变正文；实际对象服务还应检查写入完成、长度与校验值，再发布元数据引用。多工作者更新报告时，要用条件版本更新防止两次更新取得同一版本。

缓存失效和对象缺失怎样区分

缓存未命中时，程序可以重新读正文并填充缓存；对象文件缺失时，则抛 FileNotFoundError。后者是数据引用或存储问题，不能用缓存为空的正常分支掩盖。如果旧缓存仍保留正文，也要安排完整性检查，以免一直遮蔽对象缺失。

数据库中的任务状态不能只存缓存。缓存进程重启、容量淘汰或过期都可能移除记录。正文对象还需要访问控制与保留策略，数据库引用需与用户权限一起校验；公开可猜测的对象路径不应作为授权保护。

清理和恢复要按引用关系进行

更新后旧正文文件仍在，适合保留历史版本，但当前数据库只引用最新版本。如果需要完整历史，应另外建立版本表，再根据保留期清理无人引用的对象。清理前核对全部有效引用，并考虑并发写入中的暂存对象。

配套测试覆盖首次读取、缓存命中、版本更新、连接重建和缺失正文引用。本地文件不能验证云对象存储的网络超时、签名URL和一致性规则；替换存储接口后需根据服务契约补测试。SQLite事务接口可以帮助理解元数据提交和回滚的具体行为。

改一处输入，核对机制变化

将 run_demo 第二次保存的“销售额120元”改为“销售额150元”，运行 python -X utf8 storage_demo.py。前两行保持100元，后两行变为v2、150元；缓存键随版本切换。恢复120元再运行。

运行配套测试

将下面测试保存为同一文件夹的 test_storage_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_storage_demo.py

完整测试：test_storage_demo.py

import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from storage_demo import Reports


class StorageTests(unittest.TestCase):
    def setUp(self):
        self.folder = TemporaryDirectory()
        self.reports = Reports(self.folder.name)

    def tearDown(self):
        self.reports.close()
        self.folder.cleanup()

    def test_cache_and_version(self):
        # 报告更新后切换版本缓存键，不能把旧正文当作新版本。
        self.reports.save("r", "旧")
        self.assertEqual(self.reports.read("r"), ("旧", "object", 1))
        self.assertEqual(self.reports.read("r"), ("旧", "cache", 1))
        self.reports.save("r", "新")
        self.assertEqual(self.reports.read("r"), ("新", "object", 2))

    def test_restart_preserves_data(self):
        self.reports.save("r", "正文")
        self.reports.close()
        self.reports = Reports(self.folder.name)
        self.assertEqual(self.reports.read("r"), ("正文", "object", 1))

    def test_missing_reference_raises(self):
        with self.assertRaises(KeyError):
            self.reports.read("missing")
        self.reports.save("r", "正文")
        key = self.reports.db.execute("SELECT object_key FROM report").fetchone()[0]
        (Path(self.folder.name) / key).unlink()
        with self.assertRaises(FileNotFoundError):
            self.reports.read("r")


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“数据库、缓存与对象存储”，哪种做法更稳妥？

对应知识点：L4-storage

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-storage

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-storage

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-storage

你的回答

参考资料

Python sqlite3 文档：数据库连接、事务和查询。
Redis 缓存淘汰说明：理解容量受限时缓存怎样释放空间。
Amazon S3 对象存储介绍：对象、存储桶和对象键。