SQLite WAL backups: snapshots and restore tests

File: backup_sqlite.py. Save this complete file without removing lines.

from contextlib import closing
import os
from pathlib import Path
import sqlite3
import time
import uuid


def snapshot(source, destination, *, budget_seconds=10):
    source, destination = Path(source).resolve(), Path(destination).resolve()
    if source == destination or budget_seconds <= 0:
        raise ValueError('Use different paths and a positive time budget')
    if not source.is_file():
        raise FileNotFoundError(source)
    temporary = destination.with_name(destination.name + '.pending-' + uuid.uuid4().hex)
    deadline = time.monotonic() + budget_seconds
    def progress(status, remaining, total):
        if time.monotonic() > deadline:
            raise TimeoutError('Backup exceeded its time budget')
    try:
        with closing(sqlite3.connect(source.as_uri() + '?mode=ro', uri=True, timeout=.25)) as live:
            with closing(sqlite3.connect(temporary)) as backup:
                live.backup(backup, pages=32, progress=progress, sleep=.05)
                if backup.execute('PRAGMA quick_check').fetchall() != [('ok',)]:
                    raise RuntimeError('Backup integrity check failed')
        # Both paths are in one directory: atomic publication, no overwrite.
        os.link(temporary, destination)
    finally:
        temporary.unlink(missing_ok=True)
    return destination

Call snapshot with distinct source and destination paths. The source is read-only, connections are closed and hard-link publication preserves an existing destination.

File: seed_wal.py. Save this complete file without removing lines.

from contextlib import closing
from pathlib import Path
import sqlite3
import tempfile
from backup_sqlite import snapshot


with tempfile.TemporaryDirectory(prefix='liyan-wal-demo-') as folder:
    source = Path(folder) / 'live.db'
    destination = Path(folder) / 'backup.db'
    live = sqlite3.connect(source)
    try:
        assert live.execute('PRAGMA journal_mode=WAL').fetchone()[0] == 'wal'
        live.execute('CREATE TABLE documents(id INTEGER PRIMARY KEY, title TEXT NOT NULL)')
        live.executemany('INSERT INTO documents(title) VALUES (?)',
                         [(f'document-{i}',) for i in range(100)])
        live.commit()
        assert Path(str(source) + '-wal').is_file()
        snapshot(source, destination)
        with closing(sqlite3.connect(destination)) as restored:
            assert restored.execute('SELECT COUNT(*) FROM documents').fetchone()[0] == 100
            print('restored rows:', 100)
    finally:
        live.close()

python seed_wal.py should print restored rows: 100. The source connection remains open until verification, and its data is committed before copying.

File: test_backup_sqlite.py. Save this complete file without removing lines.

from contextlib import closing
from pathlib import Path
import sqlite3
import tempfile
import unittest
from backup_sqlite import snapshot


class BackupTests(unittest.TestCase):
    def test_wal_snapshot_with_open_writer_connection(self):
        with tempfile.TemporaryDirectory(prefix='liyan-backup-test-') as folder:
            source, target = Path(folder) / 'live.db', Path(folder) / 'backup.db'
            live = sqlite3.connect(source)
            try:
                live.execute('PRAGMA journal_mode=WAL')
                live.execute('CREATE TABLE documents(id INTEGER PRIMARY KEY, title TEXT)')
                live.execute('INSERT INTO documents VALUES (?, ?)', (1, 'approved'))
                live.commit()
                snapshot(source, target)
                with closing(sqlite3.connect(target)) as copied:
                    self.assertEqual(copied.execute('SELECT title FROM documents').fetchall(), [('approved',)])
                    self.assertEqual(copied.execute('PRAGMA integrity_check').fetchone(), ('ok',))
            finally:
                live.close()

    def test_missing_source_is_not_created(self):
        with tempfile.TemporaryDirectory(prefix='liyan-backup-missing-') as folder:
            source = Path(folder) / 'missing.db'
            with self.assertRaises(FileNotFoundError):
                snapshot(source, Path(folder) / 'backup.db')
            self.assertFalse(source.exists())

    def test_existing_target_survives(self):
        with tempfile.TemporaryDirectory(prefix='liyan-backup-existing-') as folder:
            source, target = Path(folder) / 'live.db', Path(folder) / 'existing.db'
            with closing(sqlite3.connect(source)) as live:
                live.execute('CREATE TABLE documents(id INTEGER PRIMARY KEY)')
            target.write_bytes(b'keep this file')
            with self.assertRaises(FileExistsError):
                snapshot(source, target)
            self.assertEqual(target.read_bytes(), b'keep this file')
            self.assertEqual(sorted(p.name for p in Path(folder).iterdir()), ['existing.db', 'live.db'])


if __name__ == '__main__':
    unittest.main()

Run python -m unittest test_backup_sqlite.py. Missing-source and existing-target cases ensure failures do not become a misleading apparent success.

File: restore_check.py. Save this complete file without removing lines.

from contextlib import closing
import argparse
import json
from pathlib import Path
import sqlite3


def verify_restore(path):
    path = Path(path).resolve()
    with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)) as database:
        integrity = database.execute('PRAGMA integrity_check').fetchall()
        foreign_keys = database.execute('PRAGMA foreign_key_check').fetchall()
        rows = database.execute('SELECT COUNT(*) FROM documents').fetchone()[0]
    if integrity != [('ok',)] or foreign_keys:
        raise RuntimeError('The restored database failed validation')
    return {'integrity': 'ok', 'foreign_key_errors': 0, 'documents': rows}


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('database')
    args = parser.parse_args()
    print(json.dumps(verify_restore(args.database)))

For an example backup, run python restore_check.py backup.db. The documents table is this example schema; define application-specific acceptance queries for real recovery.