feat(backup): snapshot SQLite cohérent et contrôle d'intégrité (#89)
Implémente les trois fonctions de gestion de snapshots SQLite: - make_snapshot: crée un snapshot gzippé cohérent via sqlite3.Connection.backup() - write_snapshot: écrit le snapshot décompressé à destination - verify_integrity: valide l'intégrité d'une base avec PRAGMA integrity_check Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,40 @@
|
|||||||
|
import gzip
|
||||||
|
import sqlite3
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
def make_snapshot(db_path: Path) -> bytes:
|
||||||
|
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as tmp:
|
||||||
|
tmp_path = Path(tmp.name)
|
||||||
|
try:
|
||||||
|
src = sqlite3.connect(str(db_path))
|
||||||
|
try:
|
||||||
|
dst = sqlite3.connect(str(tmp_path))
|
||||||
|
try:
|
||||||
|
src.backup(dst)
|
||||||
|
finally:
|
||||||
|
dst.close()
|
||||||
|
finally:
|
||||||
|
src.close()
|
||||||
|
return gzip.compress(tmp_path.read_bytes())
|
||||||
|
finally:
|
||||||
|
tmp_path.unlink(missing_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
def write_snapshot(gzipped: bytes, dest: Path) -> None:
|
||||||
|
dest.write_bytes(gzip.decompress(gzipped))
|
||||||
|
|
||||||
|
|
||||||
|
def verify_integrity(db_path: Path) -> bool:
|
||||||
|
try:
|
||||||
|
conn = sqlite3.connect(str(db_path))
|
||||||
|
except sqlite3.DatabaseError:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
row = conn.execute("PRAGMA integrity_check").fetchone()
|
||||||
|
return row is not None and row[0] == "ok"
|
||||||
|
except sqlite3.DatabaseError:
|
||||||
|
return False
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
@@ -0,0 +1,36 @@
|
|||||||
|
import sqlite3
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from src.backup.snapshot import make_snapshot, verify_integrity, write_snapshot
|
||||||
|
|
||||||
|
|
||||||
|
def _make_db(path: Path):
|
||||||
|
conn = sqlite3.connect(str(path))
|
||||||
|
conn.execute("CREATE TABLE t (id INTEGER, val TEXT)")
|
||||||
|
conn.execute("INSERT INTO t VALUES (1, 'a'), (2, 'b')")
|
||||||
|
conn.commit()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_snapshot_roundtrip_preserves_rows(tmp_path):
|
||||||
|
src = tmp_path / "users.sqlite"
|
||||||
|
_make_db(src)
|
||||||
|
gzipped = make_snapshot(src)
|
||||||
|
restored = tmp_path / "restored.sqlite"
|
||||||
|
write_snapshot(gzipped, restored)
|
||||||
|
conn = sqlite3.connect(str(restored))
|
||||||
|
rows = conn.execute("SELECT id, val FROM t ORDER BY id").fetchall()
|
||||||
|
conn.close()
|
||||||
|
assert rows == [(1, "a"), (2, "b")]
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_integrity_ok(tmp_path):
|
||||||
|
src = tmp_path / "users.sqlite"
|
||||||
|
_make_db(src)
|
||||||
|
assert verify_integrity(src) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_verify_integrity_detects_corruption(tmp_path):
|
||||||
|
bad = tmp_path / "bad.sqlite"
|
||||||
|
bad.write_bytes(b"this is not a sqlite database")
|
||||||
|
assert verify_integrity(bad) is False
|
||||||
Reference in New Issue
Block a user