File size: 3,215 Bytes
34b54ad | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """
PA/POS Database Backup Script.
Creates atomic, point-in-time backups of:
• app.db (SQLite: chats, notes, profile facts, files)
• embeddings/embeddings.db (SQLite: vector embeddings)
Uses the SQLite Online Backup API (sqlite3.backup) to avoid database lock issues.
Retains the last N backups (default: 7) and purges older archives.
"""
from __future__ import annotations
import argparse
import logging
import shutil
import sqlite3
import tarfile
from datetime import UTC, datetime
from pathlib import Path
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("backup")
ROOT_DIR = Path(__file__).resolve().parents[1]
BACKUPS_DIR = ROOT_DIR / "data" / "backups"
APP_DB = ROOT_DIR / "app.db"
EMBEDDINGS_DB = ROOT_DIR / "embeddings" / "embeddings.db"
def backup_sqlite(source_path: Path, dest_path: Path) -> bool:
"""Use SQLite Online Backup API for an atomic, non-blocking backup."""
if not source_path.exists():
logger.warning("Source database '%s' does not exist yet. Skipping.", source_path)
return False
dest_path.parent.mkdir(parents=True, exist_ok=True)
try:
src = sqlite3.connect(str(source_path))
dst = sqlite3.connect(str(dest_path))
with dst:
src.backup(dst, pages=100)
dst.close()
src.close()
logger.info("Successfully backed up '%s' -> '%s'", source_path.name, dest_path)
return True
except Exception as exc:
logger.error("Failed to backup '%s': %s", source_path, exc)
return False
def run_backup(retention: int = 7) -> Path:
"""Run full backup and rotate old archives."""
BACKUPS_DIR.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
temp_dir = BACKUPS_DIR / f"temp_{timestamp}"
temp_dir.mkdir(parents=True, exist_ok=True)
backed_up_any = False
if APP_DB.exists():
backed_up_any |= backup_sqlite(APP_DB, temp_dir / "app.db")
if EMBEDDINGS_DB.exists():
backed_up_any |= backup_sqlite(EMBEDDINGS_DB, temp_dir / "embeddings.db")
archive_path = BACKUPS_DIR / f"pabot_backup_{timestamp}.tar.gz"
with tarfile.open(archive_path, "w:gz") as tar:
for file in temp_dir.glob("*.db"):
tar.add(file, arcname=file.name)
shutil.rmtree(temp_dir, ignore_errors=True)
logger.info("Created backup archive: %s (%d bytes)", archive_path.name, archive_path.stat().st_size)
# Rotate old backups
all_backups = sorted(BACKUPS_DIR.glob("pabot_backup_*.tar.gz"), key=lambda p: p.stat().st_mtime)
if len(all_backups) > retention:
to_delete = all_backups[:-retention]
for old_file in to_delete:
old_file.unlink(missing_ok=True)
logger.info("Purged old backup: %s", old_file.name)
return archive_path
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Backup PA/POS SQLite databases")
parser.add_argument("--retention", type=int, default=7, help="Number of backups to keep (default: 7)")
args = parser.parse_args()
archive = run_backup(retention=args.retention)
print(f"Backup complete: {archive}")
|