（六）让数据库结构按版本前进

第五章介绍了迁移，第十五章用 Django 生成过迁移文件。这里手写两组迁移，观察同一件事怎样落实到 SQLite：第一版有书目表，第二版增加批次、拒收记录和来源关联。

# 项目文件：catalog_project/src/cataloghub/storage.py
from contextlib import closing
import json
from pathlib import Path
import sqlite3

SCHEMA = 2
MIGRATIONS = [
    [
        "CREATE TABLE books(source_id TEXT PRIMARY KEY, title TEXT NOT NULL, finished INTEGER NOT NULL CHECK(finished IN (0,1)), tags_json TEXT NOT NULL)"
    ],
    [
        "CREATE TABLE runs(id TEXT PRIMARY KEY, source TEXT NOT NULL, report_json TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)",
        "CREATE TABLE rejected(run_id TEXT REFERENCES runs(id), item_index INTEGER, page TEXT, reason TEXT, PRIMARY KEY(run_id,item_index))",
        "ALTER TABLE books ADD COLUMN updated_run TEXT REFERENCES runs(id)",
        "CREATE INDEX books_title ON books(title)",
    ],
]


def open_db(path, readonly=False):
    path = Path(path).resolve()
    if readonly:
        # file URI 的 mode=ro 让 SQLite 以只读方式打开既有数据库。
        connection = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=5)
    else:
        path.parent.mkdir(parents=True, exist_ok=True)
        connection = sqlite3.connect(path, timeout=5)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys=ON")
    return connection


def migrate(path):
    # 迁移语句与版本推进处在同一显式事务中。
    with closing(open_db(path)) as connection:
        connection.execute("PRAGMA journal_mode=WAL")
        try:
            connection.execute("BEGIN IMMEDIATE")
            version = connection.execute("PRAGMA user_version").fetchone()[0]
            if version > SCHEMA:
                raise ValueError("数据库版本高于当前程序")
            for index in range(version, SCHEMA):
                for statement in MIGRATIONS[index]:
                    connection.execute(statement)
                connection.execute(f"PRAGMA user_version={index + 1}")
            connection.commit()
        except Exception:
            connection.rollback()
            raise

PRAGMA user_version 保存这个应用采用的结构版本；journal_mode=WAL 让写入日志与主文件分开。迁移从当前版本继续执行，已有第一版的 books 不会被重建。第二版增加的 updated_run 对已有书目为空，新导入或更新的书目再指向对应批次。

只读连接使用 path.as_uri() 把绝对路径转成文件 URI，再加 ?mode=ro，通过 uri=True 让 SQLite 按 URI 解释。文件不存在时直接报错，查询启动不会顺手创建一份空库。

逐条使用 execute()，让迁移语句和版本更新处在显式事务中。途中出错就回退，再运行时仍从先前版本开始。foreign_keys=ON 对每条连接开启关联检查；写入书目前，必须先存在相应的批次记录。

从 user_version=1 开始，新建批次表、补上来源列、保留旧行，再提交为版本 2

# 首次准备或程序升级时执行，完成后再启动查询工作进程。
uv run catalog migrate