diff --git a/src/transcription/services/maintenance.py b/src/transcription/services/maintenance.py index 6e65924..d564d9a 100644 --- a/src/transcription/services/maintenance.py +++ b/src/transcription/services/maintenance.py @@ -7,12 +7,10 @@ from dataclasses import dataclass from datetime import UTC from datetime import datetime from pathlib import Path -from typing import Any from uuid import UUID from sqlalchemy import update from sqlalchemy.exc import SQLAlchemyError -from sqlmodel import SQLModel from sqlmodel import col from sqlmodel import func from sqlmodel import select @@ -51,10 +49,6 @@ class MaintenanceError(AppError): class MaintenanceService(ServiceBase): """Persist and execute background maintenance runs.""" - def __init__(self, session_factory: Any = None, settings: Any = None): - super().__init__(session_factory=session_factory, settings=settings) - self._maintenance_table_ready = False - async def list_runs( self, *, @@ -63,7 +57,6 @@ class MaintenanceService(ServiceBase): ) -> list[MaintenanceRun]: try: async with self._session_scope(session) as _session: - await self._ensure_runs_table(session=_session) query = ( select(MaintenanceRun) .order_by(col(MaintenanceRun.created_at).desc(), col(MaintenanceRun.id).desc()) @@ -92,14 +85,13 @@ class MaintenanceService(ServiceBase): ) try: async with self._session_scope(session) as _session: - await self._ensure_runs_table(session=_session) _session.add(run) await self._finalize(session=_session, caller_session=session, refresh=(run,)) except SQLAlchemyError as exc: raise MaintenanceError( "Maintenance run could not be queued.", category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Verify database schema access and retry.", + suggestion="Run the V6.1 schema migration, then retry.", detail=f"Failed to enqueue maintenance run: {type(exc).__name__}: {exc}", ) from exc return run @@ -107,7 +99,6 @@ class MaintenanceService(ServiceBase): async def claim_next_queued_run(self, *, session: AsyncSession | None = None) -> MaintenanceRun | None: try: async with self._session_scope(session) as _session: - await self._ensure_runs_table(session=_session) now = _utc_now_naive() queued_run_id = ( select(col(MaintenanceRun.id)) @@ -144,29 +135,6 @@ class MaintenanceService(ServiceBase): detail=f"Failed to claim queued maintenance run: {type(exc).__name__}: {exc}", ) from exc - async def _ensure_runs_table(self, *, session: AsyncSession) -> None: - if self._maintenance_table_ready: - return - table = SQLModel.metadata.tables.get("maintenance_run") - if table is None: - raise MaintenanceError( - "Maintenance storage is unavailable.", - category=ErrorCategory.INTERNAL_UNEXPECTED, - suggestion="Check database schema registration, then retry.", - detail="maintenance_run table metadata is not registered.", - ) - try: - connection = await session.connection() - await connection.run_sync(lambda sync_connection: table.create(sync_connection, checkfirst=True)) - except SQLAlchemyError as exc: - raise MaintenanceError( - "Maintenance storage is unavailable.", - category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Check database availability and schema permissions, then retry.", - detail=f"Failed to ensure maintenance_run table: {type(exc).__name__}: {exc}", - ) from exc - self._maintenance_table_ready = True - async def process_next_queued_run(self, *, session: AsyncSession | None = None) -> bool: run = await self.claim_next_queued_run(session=session) if run is None: diff --git a/tests/services/test_maintenance_service.py b/tests/services/test_maintenance_service.py index fb69ed0..5e69ea5 100644 --- a/tests/services/test_maintenance_service.py +++ b/tests/services/test_maintenance_service.py @@ -10,6 +10,8 @@ from sqlalchemy import text from transcription.config import Settings from transcription.db.models import MaintenanceJobType from transcription.db.models import MaintenanceRunStatus +from transcription.errors import ErrorCategory +from transcription.services.maintenance import MaintenanceError from transcription.services.maintenance import MaintenanceExecution from transcription.services.maintenance import MaintenanceService @@ -61,15 +63,12 @@ async def test_process_next_queued_run_persists_terminal_result( @pytest.mark.asyncio -async def test_list_runs_recreates_missing_table(default_session_factory, default_settings): +async def test_list_runs_raises_when_table_is_missing(default_session_factory, default_settings): service = MaintenanceService(session_factory=default_session_factory, settings=default_settings) async with default_session_factory() as session: await session.exec(text("DROP TABLE maintenance_run")) await session.commit() - runs = await service.list_runs(limit=10) - assert runs == [] - - created = await service.enqueue_run(job_type=MaintenanceJobType.BACKUP, triggered_by="test") - listed = await service.list_runs(limit=10) - assert any(run.id == created.id for run in listed) + with pytest.raises(MaintenanceError) as exc: + await service.list_runs(limit=10) + assert exc.value.category == ErrorCategory.INFRA_PERSISTENT diff --git a/tools/migrate_v60_to_v61.py b/tools/migrate_v60_to_v61.py new file mode 100644 index 0000000..7d3fb6b --- /dev/null +++ b/tools/migrate_v60_to_v61.py @@ -0,0 +1,108 @@ +"""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())