generated from coulomb/repo-seed
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from core_hub.db import get_sessionmaker
|
|
from core_hub.migration import import_bundle, validate_bundle
|
|
|
|
|
|
def load_bundle(path: Path) -> dict[str, Any]:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
value = json.load(handle)
|
|
if not isinstance(value, dict):
|
|
raise SystemExit("migration bundle must be a JSON object")
|
|
return value
|
|
|
|
|
|
def print_report(report: dict[str, Any]) -> None:
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
|
|
|
|
async def run_import(path: Path, dry_run: bool) -> dict[str, Any]:
|
|
bundle = load_bundle(path)
|
|
sessionmaker = get_sessionmaker()
|
|
async with sessionmaker() as session:
|
|
return await import_bundle(session, bundle, dry_run=dry_run)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate or import Core Hub migration bundles.")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
validate_parser = subparsers.add_parser("validate", help="Validate a bundle without DB access.")
|
|
validate_parser.add_argument("bundle", type=Path)
|
|
|
|
import_parser = subparsers.add_parser("import", help="Import a bundle into the configured DB.")
|
|
import_parser.add_argument("bundle", type=Path)
|
|
import_parser.add_argument("--dry-run", action="store_true")
|
|
|
|
args = parser.parse_args()
|
|
if args.command == "validate":
|
|
report = validate_bundle(load_bundle(args.bundle))
|
|
else:
|
|
report = asyncio.run(run_import(args.bundle, dry_run=args.dry_run))
|
|
print_report(report)
|
|
return 0 if report.get("ok") else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|