用 Python 备份 SQLite 数据库：WAL 快照与独立恢复验证

python3 --version
python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
python3 backup_sqlite.py --help

"""Back up one SQLite database without overwriting a destination. Python 3.10+."""
import argparse
from contextlib import closing
import os
from pathlib import Path
import sqlite3
import sys
import tempfile


def backup_database(source, destination):
    source = Path(source).resolve(strict=True)
    destination = Path(destination).absolute()
    if not source.is_file():
        raise ValueError("source must be a regular file")
    if os.path.lexists(destination):
        raise FileExistsError(f"destination already exists: {destination}")
    if not destination.parent.is_dir():
        raise ValueError("destination parent directory must exist")

    # A sibling temporary directory keeps the hard link on one filesystem.
    with tempfile.TemporaryDirectory(prefix=".sqlite-backup-", dir=destination.parent) as work:
        temporary = Path(work) / "backup.sqlite"
        with closing(sqlite3.connect(source.as_uri() + "?mode=ro", uri=True,
                                     isolation_level=None)) as src:
            with closing(sqlite3.connect(temporary, isolation_level=None)) as dst:
                src.backup(dst, pages=256, sleep=0.1)
                # Make the result a standalone file even when the source uses WAL.
                mode = dst.execute("PRAGMA journal_mode=DELETE").fetchone()[0]
                if mode.lower() != "delete":
                    raise ValueError("could not select DELETE journal mode")
                checks = dst.execute("PRAGMA quick_check").fetchall()
                if checks != [("ok",)]:
                    raise ValueError(f"quick_check failed: {checks}")
        os.chmod(temporary, 0o600)
        # Unlike replace(), link() fails if another process created the destination.
        os.link(temporary, destination)
    return destination


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("source", type=Path)
    parser.add_argument("destination", type=Path)
    args = parser.parse_args()
    try:
        destination = backup_database(args.source, args.destination)
    except (OSError, ValueError, sqlite3.Error) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("INTERRUPTED", file=sys.stderr)
        return 130
    print(f"BACKUP OK: {destination}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Keep committed rows in WAL while a separate backup process runs."""
from pathlib import Path
import sqlite3
import subprocess
import sys

folder = Path("sqlite-backup-demo")
folder.mkdir(exist_ok=True)
source = folder / "live.sqlite"
if source.exists():
    raise SystemExit("Use a fresh demo directory")
con = sqlite3.connect(source, isolation_level=None)
try:
    assert con.execute("PRAGMA journal_mode=WAL").fetchone()[0] == "wal"
    con.execute("PRAGMA wal_autocheckpoint=0")
    con.execute("CREATE TABLE jobs (id INTEGER PRIMARY KEY, status TEXT NOT NULL)")
    con.executemany("INSERT INTO jobs VALUES (?, ?)", [(1, "done"), (2, "done")])
    assert Path(str(source) + "-wal").stat().st_size > 0
    print("WAL has committed data", flush=True)
    subprocess.run([sys.executable, "backup_sqlite.py", str(source),
                    str(folder / "snapshot.sqlite")], check=True)
    con.execute("INSERT INTO jobs VALUES (?, ?)", (3, "later"))
    print("LIVE rows=3; snapshot should still have 2", flush=True)
finally:
    con.close()

"""Read the backup in a fresh directory without the source or its WAL files."""
from contextlib import closing
from pathlib import Path
import shutil
import sqlite3
import tempfile

backup = Path("sqlite-backup-demo/snapshot.sqlite").resolve(strict=True)
with tempfile.TemporaryDirectory(prefix="sqlite-restore-") as work:
    restored = Path(work) / "restored.sqlite"
    shutil.copyfile(backup, restored)
    with closing(sqlite3.connect(restored.as_uri() + "?mode=ro", uri=True,
                                 isolation_level=None)) as con:
        assert con.execute("PRAGMA quick_check").fetchall() == [("ok",)]
        assert con.execute("PRAGMA journal_mode").fetchone()[0] == "delete"
        rows = con.execute("SELECT id, status FROM jobs ORDER BY id").fetchall()
        assert rows == [(1, "done"), (2, "done")], rows
print("RESTORE OK: [(1, 'done'), (2, 'done')]")

python3 demo_wal.py
python3 verify_backup.py

WAL has committed data
BACKUP OK: /your/workdir/sqlite-backup-demo/snapshot.sqlite
LIVE rows=3; snapshot should still have 2
RESTORE OK: [(1, 'done'), (2, 'done')]

python3 backup_sqlite.py sqlite-backup-demo/live.sqlite sqlite-backup-demo/snapshot.sqlite
printf 'exit=%s\n' "$?"
python3 verify_backup.py

mkdir -p backups
python3 backup_sqlite.py sqlite-backup-demo/live.sqlite "backups/progress-$(date +%Y%m%d-%H%M%S).sqlite"