用 Python sqlite3 的 backup() 在线备份正在写入的 SQLite 数据库

问题场景

在运维或开发中，经常需要备份 SQLite 数据库。如果数据库正在被写入，直接用 cp 或文件复制命令拷贝 *.db 文件，可能得到损坏或不一致的副本。原因在于 SQLite 数据库文件在写入过程中可能处于中间状态，例如回滚日志（rollback journal）或 WAL 文件尚未合并到主数据库文件。直接复制可能只拿到部分页，导致副本无法打开或数据错乱。

SQLite 提供了在线备份 API（Online Backup API），Python 的 sqlite3 模块通过 Connection.backup() 暴露了该能力。它可以在源数据库继续被读写的情况下，将数据安全地复制到目标数据库。

最小可运行示例

下面代码演示：创建源数据库，启动一个持续写入的线程，然后用 backup() 在线复制到目标数据库，并设置 pages 参数分块复制、注册 progress 进度回调。

import sqlite3
import threading
import time
import os

SRC_DB = "source.db"
DST_DB = "backup.db"

def writer(stop_event):
    conn = sqlite3.connect(SRC_DB)
    i = 0
    try:
        while not stop_event.is_set():
            conn.execute("INSERT INTO t(id, val) VALUES (?, ?)", (i, f"value-{i}"))
            conn.commit()
            i += 1
            time.sleep(0.001)
    finally:
        conn.close()

def progress(status, remaining, total):
    # status 通常为 0（进行中）或 1（完成）
    done = total - remaining
    print(f"[进度] {done}/{total} 页, status={status}")

def main():
    # 清理旧文件
    for f in (SRC_DB, DST_DB):
        if os.path.exists(f):
            os.remove(f)

    # 初始化源数据库
    conn = sqlite3.connect(SRC_DB)
    conn.execute("CREATE TABLE t(id INTEGER PRIMARY KEY, val TEXT)")
    conn.commit()
    conn.close()

    # 启动持续写入线程
    stop_event = threading.Event()
    writer_thread = threading.Thread(target=writer, args=(stop_event,))
    writer_thread.start()

    # 让写入先跑一会儿，确保备份时数据库确实在写入
    time.sleep(0.5)

    # 在线备份
    src = sqlite3.connect(SRC_DB)
    dst = sqlite3.connect(DST_DB)
    try:
        # pages=50 表示每次复制 50 页；sleep=0.05 控制每次复制后的暂停
        src.backup(dst, pages=50, progress=progress, sleep=0.05)
    finally:
        dst.close()
        src.close()

    # 停止写入
    stop_event.set()
    writer_thread.join()

    # 验证备份完整性
    check = sqlite3.connect(DST_DB)
    try:
        result = check.execute("PRAGMA integrity_check").fetchone()
        print("integrity_check:", result[0])
        count = check.execute("SELECT COUNT(*) FROM t").fetchone()[0]
        print(f"备份库记录数: {count}")
    finally:
        check.close()

    print("备份完成")

if __name__ == "__main__":
    main()

运行方法

1. 需要 Python 3.7 或更高版本（Connection.backup() 从 3.7 开始提供）。
2. 将代码保存为 backup_demo.py。
3. 在终端执行 python backup_demo.py。
4. 观察输出中的进度信息，最后会打印 integrity_check: ok 和备份库中的记录数。

pages 参数与进度回调

backup(target, *, pages=-1, progress=None, name='main', sleep=0.250) 中：

• target：目标数据库连接，可以是文件或内存数据库。
• pages：每次复制的页数。-1 表示一次性复制所有页；正数表示分块复制。分块复制可以缩短单次持有源数据库锁的时间，更适合在线场景。
• progress：回调函数，签名为 progress(status, remaining, total)。status 是状态码，remaining 是剩余页数，total 是总页数。回调应尽量轻量，避免阻塞备份。
• sleep：每次复制一批页后休眠的秒数，默认 0.250。适当增大可以让源数据库的写入有更多机会执行，但会延长备份时间。

边界与限制

• 源数据库在备份期间可以继续读写，但如果存在长时间持有排他锁的写事务，备份可能等待或失败。建议捕获 sqlite3.OperationalError 并处理。
• 目标数据库最好为空文件或全新数据库。如果目标库已有表结构，备份可能因冲突而失败。
• 备份结果最终是一致的，但不保证是某个精确时间点的快照。如果源库在备份期间被修改，副本可能包含部分新数据，但数据库完整性不会损坏。
• 大数据库备份需要额外磁盘空间和较长时间。分块复制和 sleep 可以降低对源库写入的影响，但总耗时增加。
• 不要直接复制正在写入的 SQLite 文件，尤其是存在 WAL 或 journal 文件时。使用 Connection.backup() 是更安全的选择。

小结

Python 的 sqlite3.Connection.backup() 封装了 SQLite 在线备份 API，适合在数据库持续写入时进行安全备份。通过 pages 参数分块复制、progress 回调观察进度，可以在不中断业务的前提下获得一致的副本。注意目标库状态、锁竞争和备份耗时等边界，即可在实际场景中稳定使用。