fc5: /api/migrate blueprint — multipart upload for ingest kinds + JSON for the rest + apply-without-backup guard

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-22 09:03:24 -04:00
parent c880bf8259
commit 89224a7895
3 changed files with 270 additions and 0 deletions
@@ -23,6 +23,7 @@ def all_blueprints() -> list[Blueprint]:
    from .downloads import downloads_bp
    from .gallery import gallery_bp
    from .import_admin import import_admin_bp
+    from .migrate import migrate_bp
    from .ml_admin import ml_admin_bp
    from .platforms import platforms_bp
    from .posts import posts_bp
@@ -43,6 +44,7 @@ def all_blueprints() -> list[Blueprint]:
        showcase_bp,
        settings_bp,
        import_admin_bp,
+        migrate_bp,
        suggestions_bp,
        allowlist_bp,
        aliases_bp,
@@ -0,0 +1,141 @@
+"""FC-5: /api/migrate — trigger and poll migration runs.
+
+Ingest kinds (gs_ingest, ir_ingest) accept multipart/form-data with an
+`export_file` field. All other kinds accept JSON. Apply-without-backup
+guard rejects non-dry-run ingests unless a pre_migration-tagged backup
+exists in the last 24h (override with body.force=true).
+"""
+
+import json
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from quart import Blueprint, jsonify, request
+from sqlalchemy import select
+
+from ..extensions import get_session
+from ..models import MigrationRun
+from ..tasks.migration import run_migration
+
+migrate_bp = Blueprint("migrate", __name__, url_prefix="/api/migrate")
+
+_VALID_KINDS = frozenset({
+    "backup", "gs_ingest", "ir_ingest", "tag_apply",
+    "ml_queue", "verify", "rollback",
+})
+_INGEST_KINDS = frozenset({"gs_ingest", "ir_ingest"})
+_APPLY_KINDS = frozenset({"gs_ingest", "ir_ingest", "tag_apply", "rollback"})
+
+
+def _bad(error: str, *, status: int = 400, **extra):
+    body = {"error": error}
+    body.update(extra)
+    return jsonify(body), status
+
+
+def _has_recent_pre_migration_backup() -> bool:
+    from ..services.migrators import backup as backup_mod
+    images_root = Path("/images")
+    manifest = backup_mod.find_latest_backup(images_root, tag="pre_migration")
+    if manifest is None:
+        return False
+    created_at_str = manifest.get("created_at")
+    if not created_at_str:
+        return False
+    created_at = datetime.fromisoformat(created_at_str)
+    return (datetime.now(UTC) - created_at) < timedelta(hours=24)
+
+
+def _run_to_dict(run: MigrationRun) -> dict:
+    return {
+        "id": run.id,
+        "kind": run.kind,
+        "status": run.status,
+        "dry_run": run.dry_run,
+        "started_at": run.started_at.isoformat(),
+        "finished_at": run.finished_at.isoformat() if run.finished_at else None,
+        "counts": run.counts or {},
+        "error": run.error,
+        "metadata": run.metadata_ or {},
+    }
+
+
+@migrate_bp.route("/<kind>", methods=["POST"])
+async def create_run(kind: str):
+    if kind not in _VALID_KINDS:
+        return _bad("unknown_kind", detail=f"kind must be one of {sorted(_VALID_KINDS)}")
+
+    # Ingest kinds accept multipart/form-data; everything else takes JSON.
+    if kind in _INGEST_KINDS:
+        form = await request.form
+        files = await request.files
+        if "export_file" not in files:
+            return _bad("missing_export_file", detail="multipart export_file required")
+        export_file = files["export_file"]
+        try:
+            raw = export_file.read()
+            data = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            return _bad("invalid_export_file", detail=str(exc))
+        dry_run = str(form.get("dry_run", "false")).lower() in ("true", "1", "yes")
+        force = str(form.get("force", "false")).lower() in ("true", "1", "yes")
+        params: dict = {"data": data, "dry_run": dry_run}
+    else:
+        body = await request.get_json()
+        if body is None:
+            body = {}
+        if not isinstance(body, dict):
+            return _bad("invalid_body")
+        dry_run = bool(body.get("dry_run", False))
+        force = bool(body.get("force", False))
+        params = dict(body)
+
+    is_apply = (kind in _APPLY_KINDS) and not dry_run
+    if is_apply and not force and not _has_recent_pre_migration_backup():
+        return _bad(
+            "no_backup",
+            detail="apply action requires a pre_migration-tagged backup "
+                   "in the last 24h (or force=true).",
+        )
+
+    if kind == "backup":
+        params.setdefault("tag", "pre_migration")
+
+    async with get_session() as session:
+        run = MigrationRun(kind=kind, status="pending", dry_run=dry_run)
+        session.add(run)
+        await session.commit()
+        await session.refresh(run)
+        run_id = run.id
+
+    run_migration.delay(run_id, kind, params)
+    return jsonify({"run_id": run_id, "status": "pending"}), 202
+
+
+@migrate_bp.route("/runs/<int:run_id>", methods=["GET"])
+async def get_run(run_id: int):
+    async with get_session() as session:
+        run = (await session.execute(
+            select(MigrationRun).where(MigrationRun.id == run_id)
+        )).scalar_one_or_none()
+    if run is None:
+        return _bad("not_found", status=404)
+    return jsonify(_run_to_dict(run))
+
+
+@migrate_bp.route("/runs", methods=["GET"])
+async def list_runs():
+    try:
+        limit = int(request.args.get("limit", "10"))
+    except ValueError:
+        return _bad("invalid_limit")
+    if limit < 1 or limit > 100:
+        return _bad("invalid_limit")
+
+    async with get_session() as session:
+        rows = (await session.execute(
+            select(MigrationRun)
+            .order_by(MigrationRun.id.desc())
+            .limit(limit)
+        )).scalars().all()
+    return jsonify([_run_to_dict(r) for r in rows])
@@ -0,0 +1,127 @@
+"""FC-5: /api/migrate API tests."""
+import io
+import json
+
+import pytest
+
+import backend.app.tasks.migration  # noqa: F401 — register celery task
+
+from backend.app import create_app
+from backend.app.celery_app import celery
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def app():
+    return create_app()
+
+
+@pytest.fixture
+async def client(app):
+    async with app.test_client() as c:
+        yield c
+
+
+@pytest.mark.asyncio
+async def test_post_backup_returns_202_and_id(client, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.api.migrate.run_migration",
+        type("F", (), {"delay": lambda self, *a, **k: None})(),
+    )
+    resp = await client.post("/api/migrate/backup", json={})
+    assert resp.status_code == 202
+    body = await resp.get_json()
+    assert "run_id" in body
+    assert body["status"] == "pending"
+
+
+@pytest.mark.asyncio
+async def test_post_rejects_unknown_kind(client):
+    resp = await client.post("/api/migrate/destroy", json={})
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "unknown_kind"
+
+
+@pytest.mark.asyncio
+async def test_post_ingest_rejects_missing_file(client):
+    resp = await client.post("/api/migrate/gs_ingest", data={})
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "missing_export_file"
+
+
+@pytest.mark.asyncio
+async def test_post_ingest_accepts_multipart_file(client, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.api.migrate._has_recent_pre_migration_backup",
+        lambda: True,  # pretend backup exists
+    )
+    monkeypatch.setattr(
+        "backend.app.api.migrate.run_migration",
+        type("F", (), {"delay": lambda self, *a, **k: None})(),
+    )
+
+    export = {
+        "source_app": "gallerysubscriber", "schema_version": 1,
+        "subscriptions": [], "credentials": [],
+    }
+    data = {
+        "export_file": (io.BytesIO(json.dumps(export).encode()), "gs-export.json"),
+        "dry_run": "false",
+    }
+    resp = await client.post("/api/migrate/gs_ingest", files=data)
+    assert resp.status_code == 202
+
+
+@pytest.mark.asyncio
+async def test_post_apply_without_backup_rejected(client, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.api.migrate._has_recent_pre_migration_backup",
+        lambda: False,
+    )
+    export = {
+        "source_app": "gallerysubscriber", "schema_version": 1,
+        "subscriptions": [], "credentials": [],
+    }
+    data = {
+        "export_file": (io.BytesIO(json.dumps(export).encode()), "gs-export.json"),
+        "dry_run": "false",
+    }
+    resp = await client.post("/api/migrate/gs_ingest", files=data)
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "no_backup"
+
+
+@pytest.mark.asyncio
+async def test_post_dry_run_allowed_without_backup(client, monkeypatch):
+    monkeypatch.setattr(
+        "backend.app.api.migrate._has_recent_pre_migration_backup",
+        lambda: False,
+    )
+    monkeypatch.setattr(
+        "backend.app.api.migrate.run_migration",
+        type("F", (), {"delay": lambda self, *a, **k: None})(),
+    )
+    export = {
+        "source_app": "gallerysubscriber", "schema_version": 1,
+        "subscriptions": [], "credentials": [],
+    }
+    data = {
+        "export_file": (io.BytesIO(json.dumps(export).encode()), "gs-export.json"),
+        "dry_run": "true",
+    }
+    resp = await client.post("/api/migrate/gs_ingest", files=data)
+    assert resp.status_code == 202
+
+
+@pytest.mark.asyncio
+async def test_get_run_404_unknown(client):
+    resp = await client.get("/api/migrate/runs/9999999")
+    assert resp.status_code == 404
+
+
+def test_run_migration_task_registered():
+    assert "backend.app.tasks.migration.run_migration" in celery.tasks