generated from john/python-template
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""One-time migration for V6.1 maintenance-run storage.
|
|
|
|
This migration aligns the ``maintenance_run`` table with the V6.1 model.
|
|
|
|
Steps:
|
|
1. Create ``maintenance_run`` if absent.
|
|
2. If present, add missing non-critical columns.
|
|
3. Refuse unsafe patching when critical identity/state columns are missing.
|
|
|
|
Usage::
|
|
|
|
uv run python tools/migrate_v60_to_v61.py --dry-run
|
|
uv run python tools/migrate_v60_to_v61.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections.abc import Sequence
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy import inspect as sqlalchemy_inspect
|
|
from sqlalchemy import text
|
|
from sqlalchemy.engine import Connection
|
|
from sqlmodel import SQLModel
|
|
|
|
from transcription.config import Settings
|
|
from transcription.config import get_settings
|
|
from transcription.db import models as _models # noqa: F401 (registers metadata tables)
|
|
from transcription.db.engine import get_database_url
|
|
|
|
TABLE_NAME = "maintenance_run"
|
|
CRITICAL_COLUMNS = frozenset({"id", "job_type", "status"})
|
|
|
|
|
|
def _sync_url(settings: Settings) -> str:
|
|
return get_database_url(settings).replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "")
|
|
|
|
|
|
def _existing_columns(connection: Connection) -> set[str]:
|
|
inspector = sqlalchemy_inspect(connection)
|
|
return {column["name"] for column in inspector.get_columns(TABLE_NAME)}
|
|
|
|
|
|
def _add_missing_columns(connection: Connection, *, dry_run: bool) -> int:
|
|
table = SQLModel.metadata.tables[TABLE_NAME]
|
|
existing = _existing_columns(connection)
|
|
missing = [column for column in table.columns if column.name not in existing]
|
|
if not missing:
|
|
print(" no missing columns")
|
|
return 0
|
|
|
|
for column in missing:
|
|
type_sql = column.type.compile(dialect=connection.dialect)
|
|
statement = f'alter table "{TABLE_NAME}" add column "{column.name}" {type_sql}'
|
|
print(f" add column {column.name} {type_sql}")
|
|
if not dry_run:
|
|
connection.execute(text(statement))
|
|
return len(missing)
|
|
|
|
|
|
def migrate_maintenance_run_schema(connection: Connection, *, dry_run: bool) -> None:
|
|
inspector = sqlalchemy_inspect(connection)
|
|
if TABLE_NAME not in set(inspector.get_table_names()):
|
|
print(f" create table {TABLE_NAME}")
|
|
if not dry_run:
|
|
SQLModel.metadata.tables[TABLE_NAME].create(connection)
|
|
return
|
|
|
|
existing = _existing_columns(connection)
|
|
missing_critical = sorted(CRITICAL_COLUMNS - existing)
|
|
if missing_critical:
|
|
raise RuntimeError(
|
|
"Existing maintenance_run table is missing critical columns "
|
|
f"{missing_critical}; run export/import migration instead of patching in place."
|
|
)
|
|
|
|
_add_missing_columns(connection, dry_run=dry_run)
|
|
|
|
|
|
def migrate(*, settings: Settings, dry_run: bool) -> None:
|
|
engine = create_engine(_sync_url(settings))
|
|
try:
|
|
with engine.begin() as connection:
|
|
print("Step 1: align maintenance_run schema")
|
|
migrate_maintenance_run_schema(connection, dry_run=dry_run)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
if dry_run:
|
|
print("\nDry run: nothing was written.")
|
|
else:
|
|
print("\nDone.")
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--dry-run", action="store_true", help="Report planned changes without writing")
|
|
args = parser.parse_args(argv)
|
|
|
|
settings = get_settings()
|
|
print(f"Target: {_sync_url(settings)}")
|
|
migrate(settings=settings, dry_run=args.dry_run)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|