bvandeusenandClaude Opus 5.5
3008d29f89
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 13s
CI & Build / TypeScript typecheck (push) Successful in 52s
CI & Build / integration (push) Successful in 1m2s
CI & Build / Python tests (push) Successful in 1m54s
CI & Build / Build & push image (push) Successful in 32s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""Restore a Scribe backup into an empty instance, from the command line.
|
|
|
|
python -m scribe.restore /backups/scribe-20261015T020000Z.tar
|
|
|
|
This is how a fresh instance is brought back after losing the old one. Run it
|
|
inside the Scribe container, which already has the database settings and has
|
|
already migrated the schema on start. It reads either the scheduled backup's
|
|
tar (services/backup_file.py) or the JSON the admin download produces.
|
|
|
|
It refuses a database that already holds an account. A restore re-creates
|
|
every user, project and record from the file with new ids; into a live
|
|
instance it would add a second copy of everything rather than replace it.
|
|
(Settings → Backup & Restore still offers the upload for that case, on
|
|
purpose and with its own warning.)
|
|
|
|
Unlike the upload, nothing limits the file's size here, so this is also the
|
|
way in for a backup too big for the web upload.
|
|
|
|
Afterwards, restart Scribe: search indexes are not part of a backup and are
|
|
rebuilt from the restored records at startup.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
|
|
def load_backup(path: Path) -> dict:
|
|
"""The payload in `path`, whichever of the two backup formats it is."""
|
|
from scribe.services.backup_file import is_archive, read_archive
|
|
|
|
with open(path, "rb") as f:
|
|
if is_archive(f):
|
|
return read_archive(f)
|
|
return json.load(f)
|
|
|
|
|
|
async def _account_count() -> int:
|
|
from scribe.models import async_session
|
|
from scribe.models.user import User
|
|
|
|
async with async_session() as session:
|
|
return (await session.execute(select(func.count()).select_from(User))).scalar_one()
|
|
|
|
|
|
async def run(path: Path) -> int:
|
|
from scribe.models import engine
|
|
from scribe.services.backup import restore_full_backup
|
|
from scribe.services.backup_file import ArchiveError
|
|
|
|
try:
|
|
try:
|
|
payload = load_backup(path)
|
|
except FileNotFoundError:
|
|
print(f"There is no file at {path}.", file=sys.stderr)
|
|
return 2
|
|
except (ArchiveError, ValueError) as exc:
|
|
print(f"{path} isn't a backup this Scribe can read: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
accounts = await _account_count()
|
|
if accounts:
|
|
print(
|
|
f"This database already holds {accounts} account(s). A restore adds a second "
|
|
"copy of everything rather than replacing it, so it only runs into an empty "
|
|
"instance. Start Scribe on an empty database and run this again.",
|
|
file=sys.stderr,
|
|
)
|
|
return 3
|
|
|
|
stats = await restore_full_backup(payload)
|
|
print(json.dumps({"restored": stats}, indent=2, default=str))
|
|
print("Restored. Restart Scribe so the search indexes are rebuilt.", file=sys.stderr)
|
|
return 0
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="python -m scribe.restore",
|
|
description="Restore a Scribe backup (.tar or .json) into an empty instance.",
|
|
)
|
|
parser.add_argument("file", type=Path, help="the backup file")
|
|
args = parser.parse_args(argv)
|
|
return asyncio.run(run(args.file))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|