142 lines
4.8 KiB
Python
142 lines
4.8 KiB
Python
"""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])
|