Files
transcription/tools/export_import_migration.py
T
Jim LancasterandCopilot App 67feeb28af
Quality Gate / gate (push) Failing after 47s
Repair the pre-commit quality gate and clear the ruff backlog
The pre-commit hooks declared `language: system` with bare `ruff`/`ty`
entries, but both are uv-managed dev dependencies and are not on PATH, so every
commit failed with `Executable 'ruff' not found`. Route both through
`uv run`; keep ruff blocking and make ty advisory (verbose) until its 18
whole-project diagnostics are cleared.

With the gate working, clear `ruff check .` to zero:

- 18 auto-fixes (import sorting, blank lines, `max()` simplification,
  `with` merging, unused imports).
- Real defects: `SourceNavigation` annotated but never imported in
  sources_page; two naive `datetime.now()` calls in migration.py now use
  `datetime.now(UTC)`.
- Dead parameters removed: `source_has_photo_table` (computed, passed, never
  read), `_serialize_value(key=...)`, and unused `request` on two NiceGUI
  page handlers where the framework injects it optionally.
- Mechanical line-length wrapping and one `startswith` tuple collapse.
- `# noqa: PLR0915` / `# noqa: PLR1702` on five long UI/migration
  functions, following the convention already used in jobs_page and
  settings_page, rather than refactoring during stabilization.

Full suite green (377 tests, `-m "not external"`).

Co-authored-by: Copilot App <[email protected]>
2026-08-23 16:47:11 -05:00

122 lines
4.3 KiB
Python

"""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())