사용 중인 SQLite 파일을 backup API로 복사해 검산하기

#!/usr/bin/env python3
"""Back up only the supplied P52 synthetic fixture, without overwriting anything."""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import shutil
import sqlite3
import sys
import tempfile
import time

APPLICATION_ID = 0x50523532
MAX_BYTES = 1024 * 1024
FIXTURE_SCHEMA_SQL = (
    'CREATE TABLE orders(order_id INTEGER PRIMARY KEY, customer TEXT NOT NULL, '
    'amount_cents INTEGER NOT NULL CHECK(amount_cents >= 0))'
)
FIXTURE_COLUMNS = [
    (0, 'order_id', 'INTEGER', 0, None, 1, 0),
    (1, 'customer', 'TEXT', 1, None, 0, 0),
    (2, 'amount_cents', 'INTEGER', 1, None, 0, 0),
]
ROWS = [(1001, 'reader-a', 12000), (1002, 'reader-b', 23500),
        (1003, 'reader-c', 8000), (1004, 'reader-d', 41000),
        (1005, 'reader-e', 15500)]

class DeadlineExceeded(TimeoutError):
    """Cooperative deadline observed at a check, not an OS wall-clock cutoff."""

class DeadlineProgress:
    def __init__(self, deadline):
        self.deadline = deadline
        self.calls = 0
        self.busy_calls = 0
        self.last_status = None
        self.total_pages = 0
    def check(self):
        if time.monotonic() >= self.deadline:
            raise DeadlineExceeded('best-effort deadline exceeded; no backup published')
    def __call__(self, status, remaining, total):
        self.calls += 1
        self.last_status = status
        self.total_pages = total
        if status in (sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED):
            self.busy_calls += 1
        self.check()  # Raising exits Python's backup retry loop, including BUSY.

def sha256(path, budget):
    digest = hashlib.sha256()
    with path.open('rb') as stream:
        while True:
            budget.check()
            chunk = stream.read(65536)
            if not chunk:
                return digest.hexdigest()
            digest.update(chunk)

def new_target(path):
    """Resolve the parent only; lexists must see even a dangling leaf symlink."""
    path = Path(path).expanduser()
    parent = path.parent.resolve(strict=True)
    if not parent.is_dir():
        raise NotADirectoryError(str(parent))
    target = parent / path.name
    for candidate in (target, *(Path(str(target)+s) for s in ('-journal', '-wal', '-shm'))):
        if os.path.lexists(candidate):
            raise FileExistsError(f'refusing existing target or sidecar: {candidate}')
    return target

def verify_fixture(con):
    if con.execute('PRAGMA application_id').fetchone()[0] != APPLICATION_ID:
        raise ValueError('only the P52 synthetic fixture is allowed')
    if con.execute('PRAGMA user_version').fetchone()[0] != 1:
        raise ValueError('unsupported fixture version')
    if con.execute('PRAGMA journal_mode').fetchone()[0] != 'delete':
        raise ValueError('this exercise accepts rollback-journal DELETE mode only')
    # Exact complete allow-list, with no LIKE/prefix filtering. This fixture has
    # no internal tables, views, indexes, triggers or hidden/unchecked columns.
    objects = con.execute('SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name').fetchall()
    if objects != [('table', 'orders', 'orders', FIXTURE_SCHEMA_SQL)]:
        raise ValueError('unexpected complete schema or constraints; use the supplied fixture')
    if con.execute('PRAGMA table_xinfo(orders)').fetchall() != FIXTURE_COLUMNS:
        raise ValueError('unexpected columns; use the supplied synthetic fixture')
    rows = con.execute('SELECT order_id, customer, amount_cents FROM orders ORDER BY order_id').fetchall()
    if rows != ROWS:
        raise ValueError('fixture rows differ from the expected five synthetic rows')
    count, total = con.execute('SELECT COUNT(*), SUM(amount_cents) FROM orders').fetchone()
    keys = [row[0] for row in rows]
    integrity = [row[0] for row in con.execute('PRAGMA integrity_check')]
    if count != 5 or total != 100000 or integrity != ['ok']:
        raise ValueError('row/sum/integrity verification failed')
    return {'rows': count, 'keys': keys, 'amount_cents_sum': total,
            'integrity_check': integrity, 'journal_mode': 'delete'}

def backup_verified(source, output, *, deadline_seconds=5.0, pages=1, sleep=0.05):
    if not math.isfinite(deadline_seconds) or not 0 < deadline_seconds <= 60:
        raise ValueError('deadline must be finite and in (0, 60] seconds')
    if not isinstance(pages, int) or not 1 <= pages <= 64:
        raise ValueError('pages must be an integer from 1 to 64')
    if not math.isfinite(sleep) or not 0 < sleep <= 0.25:
        raise ValueError('sleep must be finite and in (0, 0.25] seconds')
    budget = DeadlineProgress(time.monotonic() + deadline_seconds)
    source = Path(source).expanduser().resolve(strict=True)
    if not source.is_file() or source.stat().st_size > MAX_BYTES:
        raise ValueError('source must be a regular fixture file no larger than 1 MiB')
    output = new_target(output)  # Refuses files, same path, hardlinks and symlinks.
    if output == source:
        raise ValueError('source and output must differ')
    if shutil.disk_usage(output.parent).free < 3 * source.stat().st_size + MAX_BYTES:
        raise OSError('insufficient free space for this small exercise')
    before = sha256(source, budget)
    src = dst = None
    with tempfile.TemporaryDirectory(prefix='.p52-work-', dir=output.parent) as work:
        temporary = Path(work)/'copy.db'
        try:
            # No immutable=1: ordinary SQLite locks and the journal still apply.
            src = sqlite3.connect(source.as_uri()+'?mode=ro', uri=True,
                                  timeout=min(0.05, deadline_seconds), isolation_level=None)
            src.set_progress_handler(lambda: time.monotonic() >= budget.deadline, 1000)
            src.execute('BEGIN')
            source_check = verify_fixture(src)  # Establishes the read snapshot.
            budget.check()
            dst = sqlite3.connect(temporary, timeout=min(0.05, deadline_seconds), isolation_level=None)
            dst.set_progress_handler(lambda: time.monotonic() >= budget.deadline, 1000)
            # Source connection stays OPEN throughout Connection.backup().
            src.backup(dst, pages=pages, progress=budget, sleep=sleep)
            budget.check()
            backup_check = verify_fixture(dst)
            if backup_check != source_check:
                raise ValueError('source snapshot and copied database differ')
            after = sha256(source, budget)
            if after != before:
                raise ValueError('source SHA changed; no-writer fixture precondition violated')
        finally:
            # `with connection:` commits/rolls back; it does NOT close connections.
            if dst is not None:
                dst.close()
            if src is not None:
                src.close()  # Also releases the explicit read transaction.
        budget.check()
        # Flush the completed new file before publishing; no durability guarantee.
        with temporary.open('rb') as stream:
            os.fsync(stream.fileno())
        new_target(output)  # Recheck sidecars before publication.
        budget.check()
        # Same-filesystem atomic no-clobber: if target appeared, link fails.
        # Unsupported hard links fail safely; there is no replace/copy fallback.
        os.link(temporary, output)
        # TemporaryDirectory removes only this invocation's private work directory.
    return {'status': 'ok', 'source_file': source.name, 'backup_file': output.name,
            **backup_check, 'source_sha256_before': before,
            'source_sha256_after': after, 'source_sha256_unchanged': True,
            'backup_progress': {'calls': budget.calls, 'busy_calls': budget.busy_calls,
                                'last_status': budget.last_status,
                                'total_pages': budget.total_pages},
            'scope': 'synthetic_fixture_no_concurrent_writer'}

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('source', help='synthetic source.db made by create_fixture.py')
    parser.add_argument('output', help='a NEW backup filename in an existing trusted folder')
    parser.add_argument('--deadline-seconds', type=float, default=5.0)
    args = parser.parse_args()
    try:
        result = backup_verified(args.source, args.output, deadline_seconds=args.deadline_seconds)
    except (OSError, ValueError, sqlite3.Error) as error:
        print(json.dumps({'status':'error', 'error_type':type(error).__name__, 'message':str(error)}, ensure_ascii=False), file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0

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