68 lines
2.4 KiB
Python
Executable File
68 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Create or restore an SQLite backup without silently overwriting a target."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
|
|
|
|
def _check_integrity(con: sqlite3.Connection) -> None:
|
|
result = con.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if result != "ok":
|
|
raise RuntimeError(f"SQLite integrity check failed: {result}")
|
|
|
|
|
|
def copy_database(
|
|
source: str | Path, destination: str | Path, *, replace: bool = False
|
|
) -> None:
|
|
source_path = Path(source).expanduser().resolve()
|
|
destination_path = Path(destination).expanduser().resolve()
|
|
if source_path == destination_path:
|
|
raise ValueError("source and destination must differ")
|
|
if not source_path.is_file():
|
|
raise FileNotFoundError(source_path)
|
|
if destination_path.exists() and not replace:
|
|
raise FileExistsError(
|
|
f"refusing to overwrite existing database: {destination_path}"
|
|
)
|
|
destination_path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with closing(
|
|
sqlite3.connect(f"file:{source_path}?mode=ro", uri=True)
|
|
) as source_con:
|
|
_check_integrity(source_con)
|
|
with closing(sqlite3.connect(destination_path)) as destination_con:
|
|
source_con.backup(destination_con)
|
|
destination_con.commit()
|
|
_check_integrity(destination_con)
|
|
version = destination_con.execute(
|
|
"SELECT MAX(version) FROM schema_migrations"
|
|
).fetchone()[0]
|
|
if version is None:
|
|
raise RuntimeError("backup has no schema migration version")
|
|
except sqlite3.DatabaseError as exc:
|
|
raise RuntimeError(f"SQLite backup failed: {exc}") from exc
|
|
print(f"backup complete: {source_path} -> {destination_path}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("action", choices=("backup", "restore"))
|
|
parser.add_argument("source")
|
|
parser.add_argument("destination")
|
|
parser.add_argument(
|
|
"--replace",
|
|
action="store_true",
|
|
help="explicitly allow replacing the destination",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
copy_database(args.source, args.destination, replace=args.replace)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|