Python SQLite 在线备份完整实验：WAL、一致快照与恢复校验

SQLite 开着 WAL，程序提交了新数据，复制出来的 .db 文件却查不到刚建的表。这个现象并不矛盾：提交可能已经进入 WAL，主文件尚未包含这些页面。只复制文件名最像数据库的那一个，并没有复制当时的数据库状态。

在线备份要交给理解数据库页与事务的接口。Python 标准库的 Connection.backup() 可以完成这件事；但一份能用来恢复的备份，还需要明确输出文件何时可见、失败后留下什么、怎样拒绝覆盖旧备份，以及恢复端究竟核对哪些内容。

这次做一个有限范围的工具：备份单个 SQLite 的 main 库，生成新的本地文件；目标名称已经存在就拒绝。示例使用 Python 3.12，文件发布使用同目录硬链接，适用于支持硬链接的本地文件系统。它不承担远程上传、加密或备份保留策略。两个脚本放在同一目录即可运行，没有第三方 Python 依赖。

先让不完整的文件没有正式名字

直接把正式备份名交给 sqlite3.connect()，程序还没备份完，那个文件就已经存在。另一个任务若按文件名扫描，可能提前拿走尚未完成的内容。把“文件存在”当成“备份成功”，是比复制方法更容易漏掉的问题。

工具先在目标目录创建随机临时文件，备份完成并通过结构检查以后，关闭连接，再给它建立正式名称。同目录让发布动作落在同一文件系统；os.link() 的创建动作会拒绝已存在的目标，避免“先判断不存在，再覆盖写入”的竞争窗口。最后删除临时名称，两个名称曾短暂指向同一个已完成文件。

保存为 safe_backup.py：

from contextlib import closing
from pathlib import Path
import math
import os
import sqlite3
import tempfile
import time


def backup_database(source, destination, seconds=10, on_progress=None):
    if not math.isfinite(seconds) or seconds < 0:
        raise ValueError('seconds must be finite and non-negative')
    source = Path(source).expanduser().resolve(strict=True)
    destination = Path(destination).expanduser().absolute()
    if destination.exists():
        raise FileExistsError(destination)
    if not destination.parent.is_dir():
        raise FileNotFoundError(destination.parent)
    deadline = time.monotonic() + seconds
    fd, temporary = tempfile.mkstemp(prefix='.backup-', suffix='.db',
                                     dir=destination.parent)
    os.close(fd)
    temporary = Path(temporary)

    def progress(status, remaining, total):
        if time.monotonic() >= deadline:
            raise TimeoutError('backup exceeded its cooperative time budget')
        if on_progress is not None:
            on_progress(status, remaining, total)
        if time.monotonic() >= deadline:
            raise TimeoutError('backup exceeded its cooperative time budget')

    try:
        with closing(sqlite3.connect(source.as_uri() + '?mode=ro', uri=True,
                                     timeout=0.2, autocommit=True)) as src:
            with closing(sqlite3.connect(temporary, autocommit=True)) as dst:
                src.backup(dst, pages=16, sleep=0.05, progress=progress)
                if dst.execute('PRAGMA quick_check').fetchall() != [('ok',)]:
                    raise RuntimeError('backup structure check failed')
                if dst.execute('PRAGMA foreign_key_check').fetchall():
                    raise RuntimeError('backup contains foreign key violations')
        # The temporary file is closed; publish only a completed database.
        # Same-directory hard link fails if the final name already exists.
        os.link(temporary, destination)
    finally:
        temporary.unlink(missing_ok=True)
    return destination


if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('source')
    parser.add_argument('destination')
    parser.add_argument('--seconds', type=float, default=10)
    args = parser.parse_args()
    if not math.isfinite(args.seconds) or args.seconds <= 0:
        parser.error('--seconds must be finite and positive')
    print(backup_database(args.source, args.destination, args.seconds))

这里特意没有使用 with connection: 代替关闭连接。sqlite3 的连接上下文管理器主要处理事务结束，不负责替调用者关闭连接；closing() 才表达本工具需要的资源所有权。目标连接不与其他工作共享，源连接则是专门为备份打开的只读连接。

源路径先要求真实存在，再用 URI 的 mode=ro 打开。路径写错会明确失败，不会悄悄新建一份空库，然后把空库当成“备份成功”。目标名字已经存在时也明确失败，包括误把源文件本身当目标的情况。

autocommit=True 是这里针对 Python 3.12 的明确选择：工具不需要 Python 自动维持业务事务，备份接口自行组织所需的数据库动作。它不是给所有 SQLite 应用推荐同一个事务模式，更不能拿来修改原业务连接的设置。

用可控的 WAL 内容检验恢复结果

只看到“生成了 verified.db”，证据还不够。实验在源连接保持打开时关闭自动 checkpoint，建表并写入数据，让单独复制主文件的不足能够出现。然后在备份的进度回调中，通过另一个连接提交一次转账，观察生成的数据库是否仍满足业务不变量。

这里有两张完整表：wallet 的余额使用整数作为实验单位，attachment 引用钱包并存储用于增加页面数量的载荷。它们没有任何生产数据。300 条附件只为让 pages=16 有机会产生多个步骤，不用于声称备份吞吐。

保存为 demo.py，与前一个文件放在一起：

from contextlib import closing
from pathlib import Path
import shutil
import sqlite3
import tempfile
from safe_backup import backup_database


def main():
    with tempfile.TemporaryDirectory() as folder:
        root = Path(folder)
        source = root / 'live.db'
        with closing(sqlite3.connect(source, autocommit=True)) as writer:
            writer.execute('PRAGMA journal_mode=WAL')
            writer.execute('PRAGMA wal_autocheckpoint=0')
            writer.executescript('''
                CREATE TABLE wallet (
                    id INTEGER PRIMARY KEY,
                    balance INTEGER NOT NULL CHECK(balance >= 0)
                );
                CREATE TABLE attachment (
                    id INTEGER PRIMARY KEY,
                    wallet_id INTEGER NOT NULL REFERENCES wallet(id),
                    payload BLOB NOT NULL
                );
                INSERT INTO wallet VALUES (1, 100), (2, 100);
            ''')
            writer.executemany(
                'INSERT INTO attachment(wallet_id,payload) VALUES (?,?)',
                [(1, b'x' * 2048) for _ in range(300)])
            unsafe = root / 'unsafe-copy.db'
            shutil.copyfile(source, unsafe)
            with closing(sqlite3.connect(unsafe, autocommit=True)) as raw:
                try:
                    raw.execute('SELECT COUNT(*) FROM wallet').fetchone()
                except sqlite3.OperationalError:
                    print('raw main-file copy: wallet table unavailable')
                else:
                    print('raw copy happened to contain wallet; not a consistency proof')
            changed = False
            def change_once(status, remaining, total):
                nonlocal changed
                if not changed and remaining > 0:
                    writer.execute('BEGIN IMMEDIATE')
                    writer.execute('UPDATE wallet SET balance=90 WHERE id=1')
                    writer.execute('UPDATE wallet SET balance=110 WHERE id=2')
                    writer.execute('COMMIT')
                    changed = True
            target = root / 'verified.db'
            backup_database(source, target, on_progress=change_once)
            assert changed
            with closing(sqlite3.connect(target, autocommit=True)) as restored:
                rows = restored.execute('SELECT id,balance FROM wallet ORDER BY id').fetchall()
                assert rows == [(1, 90), (2, 110)]
                assert restored.execute('SELECT COUNT(*) FROM attachment').fetchone()[0] == 300
                print('restored:', rows, 'attachments=300')
            try:
                backup_database(source, target)
            except FileExistsError:
                print('existing target: refused')
            else:
                raise AssertionError('target overwritten')
            try:
                backup_database(source, root / 'missing.db', seconds=0)
            except TimeoutError:
                assert not (root / 'missing.db').exists()
                assert not list(root.glob('.backup-*'))
                print('expired budget: no published file or temporary file')
            else:
                raise AssertionError('expired backup succeeded')
            try:
                backup_database(root / 'not-there.db', root / 'none.db')
            except FileNotFoundError:
                assert not (root / 'not-there.db').exists()
                print('missing source: refused without creating a database')
            else:
                raise AssertionError('missing source created')

if __name__ == '__main__':
    print('SQLite', sqlite3.sqlite_version)
    main()

运行：

python3 --version
python3 demo.py

在 Python 3.12.14、SQLite 3.53.1 的这次运行中，输出为：

SQLite 3.53.1
raw main-file copy: wallet table unavailable
restored: [(1, 90), (2, 110)] attachments=300
existing target: refused
expired budget: no published file or temporary file
missing source: refused without creating a database

SQLite 的运行库版本由 Python 实际链接的库决定，不能只从 Python 的小版本号推断。实验打印它，方便复现时核对。主文件复制的失败形态也不该写成所有库必然“没有表”；如果之前发生过 checkpoint，复制件可能含有较旧的数据。程序保留了这个分支，提醒读者：碰巧能查询，不代表复制过程具备一致性保证。

这个例子没有同时复制 .db、-wal、-shm 来补救。对仍在变化的三个文件依次运行复制命令，也没有自动建立共同的读取时刻。此处的备份接口读取数据库的逻辑状态，生成的目标库关闭后再按完整文件交付，才是工具选择的协议。

一致快照不等于锁定函数入口那一刻

增量备份允许其他连接继续访问源库，但不是在任意写入压力下都能保证准时完成。源数据在步骤之间变化时，备份过程可能重新开始。实验只提交一次转账，使它能继续结束；如果持续修改导致反复重启，任务可能长期没有结束条件。

因此进度中的 remaining 与 total 适合描述当前步骤，不适合作为永不回退的业务进度。不要因为百分比从 80% 退回 20%，就直接认定数据库损坏，也不要用首次总页数算出一个必然准确的结束时间。

本实验故意在一个尚未完成的回调里提交转账，最终恢复到 90 与 110。它证明这次运行生成了可查询、总额正确并包含该次修改的结果；不能由此承诺所有备份都包含“最后一条实时写入”。业务要求精确的截点时，需要另外确定事务和写入协调协议，并把该截点与备份元数据一起记录。

备份调用的方向也要保持清楚：src.backup(dst) 把源复制到目标，目标原内容会被替换。本工具只为新临时文件建立目标连接，所以不会误把已经发布的历史备份交给覆盖型接口。

时间预算为什么叫协作预算

进度回调在备份步骤之间、用户回调返回后检查单调时钟，超预算就抛出异常。这里约束的是 backup 阶段；随后 quick_check、foreign_key_check 与文件发布不在这份回调预算内。它能停止反复等待或反复复制，却不是操作系统级的强制终止。一次底层读写、锁等待或文件系统调用已经花掉的时间，不能倒退回来。

代码把源连接的等待参数和重试间隔也设得较短，让回调有机会继续检查。pages 控制每次处理的页面数，不是总页数上限；sleep 也不能被当成整个备份最多等待多久。对每一步都要求严格截止的系统，应把备份放在独立的进程生命周期中管理，退出后再核对临时产物，而不是让进度函数宣称自己能中断所有 I/O。

实验中的 seconds=0 直接调用用于确定地触发过期分支；命令行入口则拒绝非正预算与NaN、Infinity等非有限数，函数入口同样拒绝非有限数和负数，避免时钟比较失去意义。测试检查了两件事：正式文件没有发布，随机临时文件也没有残留。异常不能只打印一行然后继续输出正式文件名，否则上游会把失败当成成功。

这份工具没有自动重试整个备份。文件系统报错、校验不通过与预算不足，需要不同处理；尤其正式名称已经存在时，应先判断它来自哪次任务，而不是删除后重跑。创建正式硬链接与程序报告成功之间仍可能发生进程退出，恢复流程要允许“结果存在，但调用方没收到确认”，通过检查已有文件决定后续动作。

硬链接提供的是完整文件的名称发布，不是对断电持久性的全面承诺。磁盘缓存、文件系统和目录同步策略仍属于部署环境。若备份需要跨设备容灾，本地目录里多一个名称也不能替代独立存储副本。

恢复演练要继续走到业务检查

quick_check 检查数据库的部分内部结构；foreign_key_check 检查外键关系。两者都不能证明应用数据满足所有业务约定。因此实验又查询钱包的精确余额与附件数量，不只检查 SQL 能不能执行。

使用工具备份自己的隔离样本时，可以执行：

python3 safe_backup.py ./sample.db ./sample-backup.db --seconds 15

恢复验证用另一个连接打开备份，查询表定义、对象集合与业务不变量，不要为了“验证恢复”直接覆盖正在运行的源库。源库此刻仍在继续写时，拿两个库的当前 COUNT 硬比较，也没有共同时间语义。

一个 main 库的备份不自动包含其他 ATTACH 数据库、上传目录里的文件或应用配置。若业务对象横跨这些位置，需要定义组合恢复点。代码中的函数名虽然叫 backup_database，它的能力仍由明确的单库范围决定。

真正值得保留的是那条交付边界：读取由数据库接口完成，临时内容先校验，正式名称只代表已完成的产物，恢复端再核对业务含义。只有失败文件与成功备份在流程上分得清，定期产生一堆 .db 文件才有可能变成可用的恢复能力。