v4.10 revision to remove "legacy compatibility" code
Quality Gate / gate (push) Failing after 11s

This commit is contained in:
Jim Lancaster
2026-08-22 11:21:18 -05:00
parent cf49c3c127
commit 63c21d4a14
25 changed files with 840 additions and 301 deletions
+119
View File
@@ -0,0 +1,119 @@
"""Formal export/import migration workflow for rebuilding to current schema.
Usage examples:
1) Export current DB + uploads to a bundle:
uv run python tools/export_import_migration.py export --bundle-dir .migration-bundle
2) Import bundle into a fresh target DB + uploads root:
uv run python tools/export_import_migration.py import --bundle-dir .migration-bundle --target-db .\\data\\transcription-new.db --target-upload-dir .\\data-new
3) One-shot rebuild flow:
uv run python tools/export_import_migration.py migrate --bundle-dir .migration-bundle --target-db .\\data\\transcription-new.db --target-upload-dir .\\data-new
"""
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Final
from transcription.config import get_settings
from transcription.db.migration import MigrationPaths
from transcription.db.migration import default_sync_db_url
from transcription.db.migration import export_bundle
from transcription.db.migration import import_bundle
from transcription.db.migration import migrate_via_bundle
from transcription.db.migration import sqlite_url_from_path
DEFAULT_BUNDLE_DIR: Final[Path] = Path(".migration-bundle")
def _db_url(value: str | None, *, default_url: str) -> str:
if value is None or not value.strip():
return default_url
candidate = value.strip()
if "://" in candidate:
return candidate
return sqlite_url_from_path(Path(candidate))
def _path(value: str | None, *, default_path: Path) -> Path:
if value is None or not value.strip():
return default_path
return Path(value)
def main() -> int:
settings = get_settings()
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--bundle-dir", default=str(DEFAULT_BUNDLE_DIR), help="Directory for export bundle artifacts")
common.add_argument(
"--source-db",
default=None,
help="Source database path or SQLAlchemy URL (default: current configured database)",
)
common.add_argument(
"--source-upload-dir",
default=None,
help="Source upload directory (default: current UPLOAD_DIR setting)",
)
common.add_argument(
"--target-db",
default=None,
help="Target database path or SQLAlchemy URL (required for import/migrate if not using current DB)",
)
common.add_argument(
"--target-upload-dir",
default=None,
help="Target upload directory (required for import/migrate if not using current UPLOAD_DIR)",
)
subparsers.add_parser("export", parents=[common], help="Export DB rows and uploads into bundle")
subparsers.add_parser("import", parents=[common], help="Import bundle into target DB and uploads")
subparsers.add_parser("migrate", parents=[common], help="Run export then import in one command")
args = parser.parse_args()
bundle_dir = Path(args.bundle_dir)
source_db_url = _db_url(args.source_db, default_url=default_sync_db_url(settings))
source_upload_dir = _path(args.source_upload_dir, default_path=settings.upload_dir)
target_db_url = _db_url(args.target_db, default_url=default_sync_db_url(settings))
target_upload_dir = _path(args.target_upload_dir, default_path=settings.upload_dir)
if args.command == "export":
export_bundle(
source_db_url=source_db_url,
source_upload_dir=source_upload_dir,
bundle_dir=bundle_dir,
)
print(f"Export complete: {bundle_dir}")
return 0
if args.command == "import":
import_bundle(
target_db_url=target_db_url,
target_upload_dir=target_upload_dir,
bundle_dir=bundle_dir,
)
print(f"Import complete: db={target_db_url} uploads={target_upload_dir}")
return 0
migrate_via_bundle(
MigrationPaths(
source_db_url=source_db_url,
target_db_url=target_db_url,
source_upload_dir=source_upload_dir,
target_upload_dir=target_upload_dir,
bundle_dir=bundle_dir,
)
)
print(f"Migration complete: bundle={bundle_dir} target_db={target_db_url} target_upload_dir={target_upload_dir}")
return 0
if __name__ == "__main__":
raise SystemExit(main())