SQLite 开着 WAL，只复制 .db 文件为什么会少数据？

老K 7 0

[aru_12] 给小工具加了 SQLite，再让定时脚本每天复制一次 app.db，看起来就有备份了。但启用 WAL 后，已经提交的记录可能还没进入这个文件。这次用一笔示例订单复现：原库有订单，复制出来的库也能打开，完整性检查甚至返回 ok，查订单却是空的。

SQLite 开着 WAL，只复制 .db 文件为什么会少数据？
本地复现实验：主文件副本与备份接口快照的订单查询结果。

先把「备份正常，数据却少了」跑出来

下面的实验只使用临时目录。先建立订单表并执行 checkpoint，让表结构进入主数据库，再关闭当前连接的自动 checkpoint，提交一笔新订单。在写入连接仍打开时，只复制 .db 文件。

import shutil, sqlite3, tempfile
from contextlib import closing
from pathlib import Path

with tempfile.TemporaryDirectory() as directory:
    source = Path(directory) / "live.db"
    with closing(sqlite3.connect(source)) as db:
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("PRAGMA wal_autocheckpoint=0")
        db.execute("CREATE TABLE orders(id INTEGER PRIMARY KEY)")
        db.commit()
        db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        db.execute("INSERT INTO orders VALUES(1)")
        db.commit()

        copied = Path(directory) / "copied.db"
        shutil.copyfile(source, copied)
        with closing(sqlite3.connect(copied)) as old:
            print("原库订单数：", db.execute("SELECT COUNT(*) FROM orders").fetchone()[0])
            print("副本订单数：", old.execute("SELECT COUNT(*) FROM orders").fetchone()[0])
            print("副本检查：", old.execute("PRAGMA integrity_check").fetchone()[0])

本机输出是：原库订单数为 1，副本订单数为 0，副本检查结果为 ok。这里特意控制了 checkpoint 的时机，目的是稳定复现遗漏；业务库不需要照着关闭自动 checkpoint。

WAL 模式把修改追加到日志中，事务提交时不一定改写主数据库。checkpoint 才会把日志中的页面写回主文件。这个实验复制到了一份结构正常的旧状态，所以 integrity_check 没有报错。它能查出结构问题，不能替你证明「昨天那笔订单也在」。SQLite：WAL 与 checkpoint

趁系统还在写入时，分别复制 .db、-wal 和 -shm，也不能保证这些文件来自同一状态。手动删除正在使用的日志文件，更不是备份的准备动作。

让 SQLite 自己生成一份一致快照

小型本地工具可以直接使用 Python 标准库的 Connection.backup()。它通过 SQLite 的 Online Backup API 读取数据库，而不是把主文件当普通字节文件复制，支持分步复制和其他连接访问数据库。Python backup() 文档 · SQLite Online Backup API

下面是实验包里的核心实现。每次备份使用一个尚不存在的目录，先写临时文件，关闭连接并完成检查后，才把文件命名为 snapshot.db。本次失败就清理本次目录；已有备份目录会直接报错，避免误覆盖历史快照。

"""SQLite online backup demo. Uses only Python's standard library."""
import os
import shutil
import sqlite3
import time
from contextlib import closing
from pathlib import Path


def check_database(path):
    """Check structural integrity and foreign keys in a closed snapshot."""
    uri = Path(path).resolve().as_uri() + '?mode=ro'
    with closing(sqlite3.connect(uri, uri=True)) as db:
        if db.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
            raise ValueError('integrity_check failed')
        if db.execute('PRAGMA foreign_key_check').fetchone() is not None:
            raise ValueError('foreign_key_check failed')


def make_backup(source, output_dir, *, timeout=5):
    """Create output_dir/snapshot.db; an existing directory is never reused.

    timeout is a cooperative deadline for the backup loop, not a hard limit
    for file I/O or the subsequent integrity checks. Local filesystem only.
    The output directory's parent must already exist.
    """
    source = Path(source).resolve(strict=True)
    output_dir = Path(output_dir).absolute()
    output_dir.mkdir()  # Exclusive reservation; never delete someone else's dir.
    partial = output_dir / 'snapshot.partial'
    final = output_dir / 'snapshot.db'
    deadline = time.monotonic() + timeout

    def progress(status, remaining, total):
        if time.monotonic() >= deadline:
            raise TimeoutError('backup deadline exceeded')

    try:
        uri = source.as_uri() + '?mode=ro'
        with closing(sqlite3.connect(uri, uri=True, timeout=1)) as src:
            with closing(sqlite3.connect(partial)) as dst:
                src.backup(dst, pages=64, progress=progress, sleep=0.01)
                mode = dst.execute('PRAGMA journal_mode=DELETE').fetchone()[0]
                if mode != 'delete':
                    raise ValueError('snapshot could not leave WAL mode')
        check_database(partial)
        os.replace(partial, final)
        return final
    except BaseException:
        shutil.rmtree(output_dir)
        raise

mode=ro 可以防止源路径拼错时创建一份空数据库。它不等于文件系统完全没有写入：读取 WAL 库仍可能需要共享内存文件和相应的目录权限。目标文件在连接关闭前切回 DELETE 日志模式，演练时只取这份关闭后的 snapshot.db，不依赖原库的 WAL。

pages=64 表示分步复制的页数；sleep=0.01 是遇到 busy/locked 时的等待参数，不是每复制一批就主动休眠。持续写入可能让备份反复重试，所以进度回调还检查了期限。这里的 5 秒是协作式期限，不能强制中断一次底层 I/O，也不包含后续完整性检查的耗时。调度器需要硬上限时，应另外管理备份子进程。

恢复验证要查业务数据，也要试一次写入

只检查文件存在和大小不够。实验先检查数据库结构，再单独执行 foreign_key_check；后者用于发现外键引用问题，不能被 integrity_check 替代。SQLite PRAGMA 检查说明

接着，把快照复制到全新的文件路径，重新连接，确认示例订单存在，再插入第二笔订单并提交。恢复库变成两笔，原库仍然只有一笔，证明恢复演练没有写回源库。

live orders: 1
plain copy orders: 0
plain copy integrity_check: ok
backup orders: 1
restored orders: 1
restored orders after a new write: 2
live orders after restore drill: 1

这组订单数核对针对的是本例固定数据。真实系统应使用备份时能确认的订单编号、导入批次等业务锚点做检查；业务仍在持续写入时，不能拿恢复库的数量和几分钟后的线上数量直接比较，就断定备份丢了数据。

下载实验包，自己重跑一次

下载 SQLite WAL 备份与恢复实验包（ZIP）。包内包含 backup.py、demo.py、测试和中文说明，无须安装第三方依赖。解压后进入目录运行：

python3 demo.py
python3 -m unittest -v

# 备份自己的库：源文件必须存在，输出目录必须是新目录
python3 backup.py /absolute/path/app.db /absolute/path/backup-20261115

本次在 macOS 的 Python 3.9.6 / SQLite 3.51.0，以及 Python 3.12.13 / SQLite 3.54.1 上分别跑了 9 项用例，全部通过。除主文件复制遗漏和恢复后的读写外，还覆盖了未提交记录不进入备份、源文件不存在、已有目录拒绝覆盖、超时清理、外键违规和损坏文件。另一个分步复制用例在两步之间让另一连接提交订单，检查复制完成后的结果。

接入定时任务时，再补上最后几步

备份目录使用时间戳或唯一编号，每次保留独立的一代。只有检查和恢复演练通过，才把它计入可用备份；过期清理要基于已有成功备份，而不是先删旧文件再开始复制。备份与原库放在同一块磁盘，解决不了整盘损坏，后面还需要异机或异地副本。

这份实现适合本地文件系统上的单个 SQLite 主库，没有验证网络盘、多个 ATTACH 数据库的整体一致性，也没有实现断电时的文件与目录持久化保证。正式恢复时，应先停写并关闭业务连接，再按应用的切换流程操作；演练继续使用独立路径，别直接覆盖运行中的数据库。

上一篇事务发件箱实验验证了发送进程退出之后，待发事件还能从数据库里读回来。这篇把后半步补上：备份文件里是否也有那些记录，恢复出来之后能不能继续处理业务，都要实际查一次。

🎨 原创不易，支持请点赞、转载请注明本文作者为老K

发表评论 取消回复
OwO 图片 链接 代码

分享