（十八）综合示例：持久化快照与 SQLite 书籍库
1. 任务与模块衔接
任务（自拟）：书籍记录继续保存 title、finished、tags。一方面，为第四章的 BookShelf 提供 JSON 文件存储器，让重新启动后的程序能读到已保存快照；另一方面，建立 SQLite 书籍库，支持 JSON/CSV 导入、按标签查询、标记已读、记录阅读时长和导出书目。
同名书导入时更新完成状态与标签，保留原编号及其阅读日志；同一份导入文件中的重复标题报告为记录错误。标签按集合意义保存，导出时按名称排序。导出字段与导入字段相同，阅读日志保存在数据库的日志表中。
目录名为 persistence_project。bookdata/models.py、interfaces.py、shelf.py 沿用第四章对应的定义，相对导入仍按同一个包组织；配套目录已经包含它们。这里继续实现转换、文件存储、数据库操作与启动入口。
persistence_project/
├── bookdata/
│ ├── __init__.py
│ ├── __main__.py
│ ├── models.py / interfaces.py / shelf.py
│ ├── errors.py / records.py
│ ├── atomic.py / json_io.py / csv_io.py / json_store.py
│ ├── database.py / library.py
│ └── arguments.py / cli.py
├── make_sample.py
├── save_snapshot.py
└── read_snapshot.py其中 atomic.py 使用本章“完整快照与原子替换”中的实现。目录图中用斜杠并列的是同级文件，实际文件都可在 项目目录索引 中打开。
2. 记录检查先于写入
# 项目文件：persistence_project/bookdata/errors.py
# 格式与业务错误用同一类别交给入口反馈，底层原因可以通过异常链继续查看。
class RecordError(ValueError):
pass# 项目文件：persistence_project/bookdata/records.py
# 基本对象先校验，再建立 Book；不把非空字符串强制转成 True。
from .errors import RecordError
from .models import Book
def book_from_dict(record: object) -> Book:
if not isinstance(record, dict):
raise RecordError("每条书籍记录应是对象")
title = record.get("title")
if not isinstance(title, str) or not title.strip():
raise RecordError("书名应是非空文本")
finished = record.get("finished", False)
if type(finished) is not bool:
raise RecordError("finished 应是布尔值")
tags = record.get("tags", [])
if not isinstance(tags, list) or not all(
isinstance(tag, str) and tag.strip() for tag in tags
):
raise RecordError("tags 应是非空标签文本组成的列表")
# dict.fromkeys 去重并保留首次出现顺序，返回一份独立标签列表。
unique_tags = list(dict.fromkeys(tag.strip() for tag in tags))
return Book(title.strip(), finished, unique_tags)
def validate_books(records: object) -> list[Book]:
if not isinstance(records, list):
raise RecordError("books 应是记录列表")
books = []
seen = set()
for number, record in enumerate(records, start=1):
try:
book = book_from_dict(record)
if book.title in seen:
raise RecordError("标题重复")
except RecordError as cause:
raise RecordError(f"第 {number} 条记录：{cause}") from cause
seen.add(book.title)
books.append(book)
return books先取得完整的 Book 列表，再开始导入事务，因此后面的格式错误不会让前几条记录先保存。数据库写入时仍有自己的约束；格式校验与事务回滚分别处理两种不同的失败。
3. JSON 读写与文件存储器
JSON 文档外层保存结构版本与书目列表，例如 {"version": 1, "books": [...]}。读取把文本还原为基本对象，再交给上面的检查函数：
# 项目文件：persistence_project/bookdata/json_io.py
# JSON 语法失败保留行列位置；读写都经过同一份书籍结构检查。
from dataclasses import asdict
import json
from pathlib import Path
from .atomic import write_text_atomic
from .errors import RecordError
from .models import Book
from .records import validate_books
def read_json(path: Path) -> list[Book]:
with path.open(encoding="utf-8-sig") as stream:
try:
document = json.load(stream)
except json.JSONDecodeError as cause:
raise RecordError(f"JSON 第 {cause.lineno} 行、第 {cause.colno} 列格式错误") from cause
if not isinstance(document, dict):
raise RecordError("JSON 外层应是对象")
version = document.get("version")
if type(version) is not int or version != 1:
raise RecordError("书目格式版本应为 1")
return validate_books(document.get("books"))
def write_json(path: Path, books: list[Book]) -> None:
# asdict 把数据类字段转换成记录字典，转换与校验都在替换文件之前完成。
checked = validate_books([asdict(book) for book in books])
document = {"version": 1, "books": [asdict(book) for book in checked]}
text = json.dumps(document, ensure_ascii=False, indent=2, allow_nan=False)
write_text_atomic(path, text + "\n")读取的 utf-8-sig 同时接受普通 UTF-8 和带 UTF-8 BOM 的文本，写出仍使用普通 UTF-8。JSON 错误的 lineno 与 colno 指出语法位置，业务记录错误则指出第几条记录。
# 项目文件：persistence_project/bookdata/json_store.py
# 通过结构相同的 load/save 接口接到第四章的书架，首次没有文件时建立空记录。
from pathlib import Path
from .json_io import read_json, write_json
from .models import Book
class JsonBookStore:
def __init__(self, path: Path):
self.path = path
def load(self) -> list[Book]:
try:
return read_json(self.path)
except FileNotFoundError:
return []
def save(self, books: list[Book]) -> None:
write_json(self.path, books)这个类不用修改 BookShelf。load() 每次从文件建立新 Book，save() 把记录写成独立文本；双方不共享上一章存储器中的可变列表。
4. CSV 的列检查与往返转换
# 项目文件：persistence_project/bookdata/csv_io.py
# CSV 处理外层三列，JSON 处理 tags 单元格中的数组。
import csv
from dataclasses import asdict
from io import StringIO
import json
from pathlib import Path
from .atomic import write_text_atomic
from .errors import RecordError
from .models import Book
from .records import validate_books
FIELDS = ["title", "finished", "tags"]
def read_csv(path: Path) -> list[Book]:
records = []
with path.open(encoding="utf-8-sig", newline="") as stream:
reader = csv.DictReader(stream, strict=True)
try:
headers = reader.fieldnames
except csv.Error as cause:
raise RecordError("CSV 表头结构错误") from cause
if headers is None or len(headers) != 3 or set(headers) != set(FIELDS):
raise RecordError("CSV 表头应包含 title、finished、tags")
try:
for number, row in enumerate(reader, start=1):
if None in row or any(value is None for value in row.values()):
raise RecordError(f"CSV 第 {number} 条记录列数错误")
flag = row["finished"].strip()
if flag not in {"0", "1"}:
raise RecordError(f"CSV 第 {number} 条记录的 finished 应为 0 或 1")
try:
tags = json.loads(row["tags"] or "[]")
except json.JSONDecodeError as cause:
raise RecordError(f"CSV 第 {number} 条记录的 tags 不是 JSON 数组") from cause
records.append({"title": row["title"], "finished": flag == "1", "tags": tags})
except csv.Error as cause:
raise RecordError("CSV 引号或分隔结构错误") from cause
return validate_books(records)
def write_csv(path: Path, books: list[Book]) -> None:
checked = validate_books([asdict(book) for book in books])
stream = StringIO(newline="")
writer = csv.DictWriter(stream, fieldnames=FIELDS, lineterminator="\n")
writer.writeheader()
for book in checked:
writer.writerow({
"title": book.title,
"finished": "1" if book.finished else "0",
"tags": json.dumps(book.tags, ensure_ascii=False),
})
write_text_atomic(path, stream.getvalue())表头按名称检查，允许列顺序不同；每条记录的列数、布尔文本和标签数组分别检查。CSV 的 strict=True 让结构错误交回解析器；这不代替字段校验。写出时先在 StringIO 形成完整文本，再按原子替换函数保存。
5. 数据库连接与表结构
# 项目文件：persistence_project/bookdata/database.py
# 先在事务外开启外键，再建立表和索引；初始化失败时关闭已取得的连接。
from pathlib import Path
import sqlite3
SCHEMA = """
CREATE TABLE IF NOT EXISTS books (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL UNIQUE CHECK(length(trim(title)) > 0),
finished INTEGER NOT NULL DEFAULT 0 CHECK(finished IN (0, 1))
);
CREATE TABLE IF NOT EXISTS book_tags (
book_id INTEGER NOT NULL REFERENCES books(id) ON DELETE CASCADE,
tag TEXT NOT NULL CHECK(length(trim(tag)) > 0),
PRIMARY KEY(book_id, tag)
);
CREATE TABLE IF NOT EXISTS reading_log (
id INTEGER PRIMARY KEY,
book_id INTEGER NOT NULL REFERENCES books(id) ON DELETE CASCADE,
day TEXT NOT NULL,
minutes INTEGER NOT NULL CHECK(typeof(minutes) = 'integer' AND minutes >= 0)
);
CREATE INDEX IF NOT EXISTS idx_log_book_day ON reading_log(book_id, day);
CREATE INDEX IF NOT EXISTS idx_tag_book ON book_tags(tag, book_id);
"""
def connect_database(path: Path) -> sqlite3.Connection:
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path, autocommit=True)
try:
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
connection.autocommit = False
with connection:
# executescript 执行程序内定义的多条建表语句。
connection.executescript(SCHEMA)
return connection
except sqlite3.Error:
connection.close()
raiseCREATE ... IF NOT EXISTS 接受这些表已经建立，因此重新启动可以继续用原来的数据。它不负责把旧表变成另一种新结构；字段变化仍按前一节的迁移过程安排。
6. 数据库对象与整批写入
Library 持有一份连接，离开上下文时关闭它。下面三段按顺序写进同一个 library.py；后两段的方法接在同一 class 内，保持四格缩进。示例索引中的链接打开的是组合后的完整文件。
# 项目文件：persistence_project/bookdata/library.py
# 对象负责连接生命周期；每个业务方法再明确自己的事务范围。
from collections import defaultdict
from datetime import date
from pathlib import Path
import sqlite3
from .database import connect_database
from .errors import RecordError
from .models import Book
class Library:
def __init__(self, path: Path):
self.connection = connect_database(path)
def __enter__(self):
return self
def __exit__(self, exc_type, exc, traceback):
self.connection.close()
return False先补上导入、标记与阅读日志操作：
# 续写同一文件：persistence_project/bookdata/library.py
def import_books(self, books: list[Book]) -> int:
with self.connection:
for book in books:
# 同名书执行更新，保留原编号；excluded 指这次待写入的值。
self.connection.execute("""
INSERT INTO books(title, finished) VALUES (?, ?)
ON CONFLICT(title) DO UPDATE SET finished = excluded.finished
""", (book.title, int(book.finished)))
book_id = self.connection.execute(
"SELECT id FROM books WHERE title = ?", (book.title,)
).fetchone()[0]
self.connection.execute("DELETE FROM book_tags WHERE book_id = ?", (book_id,))
self.connection.executemany(
"INSERT INTO book_tags(book_id, tag) VALUES (?, ?)",
((book_id, tag) for tag in book.tags),
)
return len(books)
def finish(self, title: str) -> None:
with self.connection:
cursor = self.connection.execute("UPDATE books SET finished = 1 WHERE title = ?", (title,))
if cursor.rowcount == 0:
raise RecordError(f"没有找到 {title}")
def record_reading(self, title: str, day: date, minutes: int) -> None:
if minutes < 0:
raise RecordError("阅读分钟数不能为负数")
with self.connection:
row = self.connection.execute("SELECT id FROM books WHERE title = ?", (title,)).fetchone()
if row is None:
raise RecordError(f"没有找到 {title}")
self.connection.execute(
"INSERT INTO reading_log(book_id, day, minutes) VALUES (?, ?, ?)",
(row[0], day.isoformat(), minutes),
)ON CONFLICT(title) DO UPDATE 在唯一标题已存在时更新该行，excluded.finished 是本次想写入的新状态。它继续使用原来的 id，所以日志仍指向同一本书。一次导入的所有书籍、标签修改放在同一事务内，失败传出后一起回滚。SQLite UPSERT
7. 查询、汇总与导出对象
继续在同一类中加入读取方法：
# 续写同一文件：persistence_project/bookdata/library.py
# 用 EXISTS 筛选标签，阅读日志仍只按书籍汇总，避免连接标签后重复累计分钟数。
def search(self, tag: str | None = None) -> list[sqlite3.Row]:
with self.connection:
return self.connection.execute("""
SELECT b.id, b.title, b.finished,
COALESCE(SUM(l.minutes), 0) AS total
FROM books AS b
LEFT JOIN reading_log AS l ON l.book_id = b.id
WHERE (? IS NULL OR EXISTS (
SELECT 1 FROM book_tags AS t
WHERE t.book_id = b.id AND t.tag = ?
))
GROUP BY b.id, b.title, b.finished
ORDER BY b.id
""", (tag, tag)).fetchall()
def export_books(self) -> list[Book]:
with self.connection:
tags = defaultdict(list)
for row in self.connection.execute("SELECT book_id, tag FROM book_tags ORDER BY book_id, tag"):
tags[row["book_id"]].append(row["tag"])
return [
Book(row["title"], bool(row["finished"]), tags[row["id"]])
for row in self.connection.execute("SELECT id, title, finished FROM books ORDER BY id")
]查询只把日志接入求和，标签用 EXISTS 筛选，所以两个标签不会让阅读分钟数重复出现。fetchall 在事务离开前取得完整结果，外面显示时不再继续依赖那份游标。
导出用两次查询分别取得标签与书目，再组装 Book。没有在每本书上重新发一次标签查询，也没有把两组一对多关系一次连接后再错误求和。
8. 启动参数与错误反馈
# 项目文件：persistence_project/bookdata/arguments.py
# add_subparsers 为每个动作建立自己的参数集合，required=True 要求选择动作。
import argparse
from datetime import date
from pathlib import Path
def build_parser():
parser = argparse.ArgumentParser(description="SQLite 书籍库")
parser.add_argument("--db", type=Path, default=Path("data/library.sqlite3"))
actions = parser.add_subparsers(dest="command", required=True)
for name in ("import-json", "import-csv", "export-json", "export-csv"):
action = actions.add_parser(name)
action.add_argument("path", type=Path)
listing = actions.add_parser("list")
listing.add_argument("--tag")
finish = actions.add_parser("finish")
finish.add_argument("title")
log = actions.add_parser("log")
log.add_argument("title")
log.add_argument("day", type=date.fromisoformat)
log.add_argument("minutes", type=int)
return parser--db 是主解析器的选项，写在具体动作之前；list --tag 基础 则是 list 自己的选项。type=Path、type=date.fromisoformat 都把前面学过的可调用对象交给解析器，让它实际完成文本转换。
# 项目文件：persistence_project/bookdata/cli.py
# 入口协调转换、数据库操作和输出，错误反馈在资源关闭后执行。
import sqlite3
from .arguments import build_parser
from .csv_io import read_csv, write_csv
from .json_io import read_json, write_json
from .library import Library
def main():
parser = build_parser()
args = parser.parse_args()
try:
with Library(args.db) as library:
if args.command in {"import-json", "import-csv"}:
reader = read_json if args.command == "import-json" else read_csv
books = reader(args.path)
print("已导入", library.import_books(books), "本")
elif args.command in {"export-json", "export-csv"}:
writer = write_json if args.command == "export-json" else write_csv
books = library.export_books()
writer(args.path, books)
print("已导出", len(books), "本")
elif args.command == "finish":
library.finish(args.title)
print("已标记", args.title)
elif args.command == "log":
library.record_reading(args.title, args.day, args.minutes)
print("已记录", args.minutes, "分钟")
else:
rows = library.search(args.tag)
if not rows:
print("书目为空")
for row in rows:
status = "已读" if row["finished"] else "未读"
print(row["id"], row["title"], status, f"{row['total']}分钟")
except (OSError, ValueError, sqlite3.Error) as error:
parser.exit(1, f"错误：{error}\n")parser.exit(1, message) 在标准错误输出显示说明，以非零状态结束这次启动；参数解析本身的格式错误则由 argparse 处理。这里只捕获已能给出有用说明的文件、记录与数据库错误，其余程序错误仍显示自己的调用位置，便于下一章继续排查。
# 项目文件：persistence_project/bookdata/__init__.py
# 包初始化不创建数据库或读写文件，具体操作由入口选择。
"""文件格式与 SQLite 书籍库。"""# 项目文件：persistence_project/bookdata/__main__.py
# 沿用第三章的包入口写法。
from .cli import main
if __name__ == "__main__":
main()9. 两次独立启动之间保存快照
# 项目文件：persistence_project/save_snapshot.py
# 书架对象沿用第四章的实现，只把存储器换成 JSON 文件。
from pathlib import Path
from bookdata.json_store import JsonBookStore
from bookdata.shelf import BookShelf
shelf = BookShelf(JsonBookStore(Path("data/shelf.json")))
shelf.add("Python")
shelf.finish("Python")
shelf.add_tag("Python", "基础")
shelf.save()
print("已保存", len(shelf), "本")# 项目文件：persistence_project/read_snapshot.py
# 新进程建立新对象，从文件取得之前保存的记录。
from pathlib import Path
from bookdata.json_store import JsonBookStore
from bookdata.shelf import BookShelf
shelf = BookShelf(JsonBookStore(Path("data/shelf.json")))
for book in shelf:
print(book.title, book.finished, book.tags)在项目目录先运行 python save_snapshot.py，第一次得到“已保存 1 本”；结束后再运行 python read_snapshot.py，得到 Python True ['基础']。对象不再是同一份，状态通过磁盘中的文本延续下来。
10. 准备书目、导入、记录与导出
# 项目文件：persistence_project/make_sample.py
# 生成相同内容的 JSON 与 CSV，便于比较两种导入路线。
from pathlib import Path
from bookdata.csv_io import write_csv
from bookdata.json_io import write_json
from bookdata.models import Book
books = [Book("Python", False, ["基础", "语法"]), Book("NumPy", True, ["数值"])]
write_json(Path("data/books.json"), books)
write_csv(Path("data/books.csv"), books)
print("已准备", len(books), "本")在 persistence_project 目录执行：
# 先形成输入资料，再用数据库文件保存导入结果。
python make_sample.py
python -m bookdata import-json data/books.json
# 每条命令都开启一次新进程，读取同一份数据库。
python -m bookdata log Python 2026-10-19 20
python -m bookdata log Python 2026-10-20 35
python -m bookdata log NumPy 2026-10-20 45
python -m bookdata finish Python
python -m bookdata list
python -m bookdata list --tag 基础
# 导出书目到两个独立文件，再以 CSV 导入另一份库比较记录。
python -m bookdata export-json data/export.json
python -m bookdata export-csv data/export.csv
python -m bookdata --db data/from-csv.sqlite3 import-csv data/export.csv完整查询得到：
1 Python 已读 55分钟
2 NumPy 已读 45分钟按基础标签查询只留下第一行。导入另一份库的是书目字段，因此新库的两本书、完成状态和标签一致；这些阅读时长仍保存在原库日志表中。
11. 失败输入与工作量
把输入文件第二本书的 finished 改成字符串 "true"，入口应报告第 2 条记录的字段错误，数据库已有书目保持原状。再试语法损坏的 JSON、缺列的 CSV、重复标题、未知书名和负分钟数，分别定位是解析、记录校验还是数据库操作拒绝了这次处理。
JSON/CSV 导入先保存完整的转换结果，再在一个事务中写入，Python 侧空间随书目和标签总量增长。要处理更大的来源，可以按批转换与导入，但每批提交也会把失败撤回范围改成当前批，任务的“一起成功或一起撤回”含义要跟着调整。
这份项目完整覆盖了数据来源、转换、事务与输出。下一章把这些检查固定成可重复运行的测试，并观察错误输入在实际调用中的第一处状态差异。