Function upsert_run

core/db.py:129–167

Insert or update a run record.

(run_id: str, data: dict)

# ── Run CRUD ───────────────────────────────────────────────────────────────────

def upsert_run(run_id: str, data: dict) -> None:
    """Insert or update a run record."""
    cols = [
        "id", "engagement_id", "operator", "target", "mode",
        "started_at", "completed_at", "risk_score", "risk_color",
        "total_hosts", "total_services", "total_cves",
        "confirmed_exploitable", "already_exploited",
        "remediations_generated", "remediations_applied", "verified_closed",
        "errors", "report_path",
    ]
    vals = [
        run_id,
        data.get("engagement_id", ""),
        data.get("operator", ""),
        data.get("target", ""),
        data.get("mode", ""),
        data.get("started_at", ""),
        data.get("completed_at"),
        data.get("risk_score", 0),
        data.get("risk_color", "UNKNOWN"),
        data.get("total_hosts", 0),
        data.get("total_services", 0),
        data.get("total_cves", 0),
        data.get("confirmed_exploitable", 0),
        data.get("already_exploited", 0),
        data.get("remediations_generated", 0),
        data.get("remediations_applied", 0),
        data.get("verified_closed", 0),
        json.dumps(data.get("errors", [])),
        data.get("report_path"),
    ]
    placeholders = ", ".join(["?"] * len(cols))
    updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "id")
    sql = (
        f"INSERT INTO runs ({', '.join(cols)}) VALUES ({placeholders}) "
        f"ON CONFLICT(id) DO UPDATE SET {updates}"
    )
    with get_db() as conn:
        conn.execute(sql, vals)


def get_run(run_id: str) -> dict | None: