import asyncio import json import time from contextlib import asynccontextmanager from datetime import datetime from pathlib import Path from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from config import settings, BASE_DIR from database import ( init_db, get_instances, get_instance, add_instance, update_instance, delete_instance, set_cache, get_cache ) from veeam_client import VeeamClient, port_for_version, detect_version_from_host _cache: dict[str, tuple[float, dict]] = {} def _cached(key: str, ttl: int = None) -> dict | None: ttl = ttl or settings.cache_ttl entry = _cache.get(key) if entry and (time.time() - entry[0]) < ttl: return entry[1] return None def _set_cache_inmem(key: str, data: dict): _cache[key] = (time.time(), data) # ─── Pydantic models ─────────────────────────────────────────────── class InstanceCreate(BaseModel): name: str host: str port: int | None = None username: str password: str version: str = "auto" notes: str = "" class InstanceUpdate(BaseModel): name: str | None = None host: str | None = None port: int | None = None username: str | None = None password: str | None = None version: str | None = None is_active: bool | None = None notes: str | None = None class InstanceResponse(BaseModel): id: int name: str host: str port: int username: str version: str notes: str is_active: bool created_at: str updated_at: str class DashboardResponse(BaseModel): instances: list summary: dict updated_at: str # ─── App lifecycle ───────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): await init_db() yield app = FastAPI(title=settings.app_name, version="1.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ─── Helpers ─────────────────────────────────────────────────────── def _serialize_instance(inst) -> dict: return { "id": inst.id, "name": inst.name, "host": inst.host, "port": inst.port, "username": inst.username, "version": inst.version, "notes": inst.notes, "is_active": inst.is_active, "created_at": inst.created_at.isoformat() if inst.created_at else "", "updated_at": inst.updated_at.isoformat() if inst.updated_at else "", } def _safe_str(v) -> str: if v is None: return "" if isinstance(v, dict): return str(v.get("result") or v.get("Result") or "unknown") return str(v) async def _collect_instance(inst) -> dict | None: port = inst.port or port_for_version(inst.version) ver = inst.version if ver == "auto": ver = detect_version_from_host(inst.host) port = port_for_version(ver) client = VeeamClient(inst.host, port, inst.username, inst.password, ver) try: role = await client.connect() jobs_data = [] sessions_data = [] repos_data = [] repo_states = [] alarms_data = [] agents_data = [] errors = [] try: jobs_data = await client.fetch_jobs() except Exception as e: errors.append(f"jobs: {e}") jobs_data = [] try: sessions_data = await client.fetch_sessions(100) except Exception as e: errors.append(f"sessions: {e}") sessions_data = [] if not jobs_data and sessions_data: job_map: dict[str, dict] = {} for s in sessions_data: name = s.get("jobName") or s.get("name") or s.get("Name") or "" if not name: continue ts = s.get("endTime") or s.get("creationTime") or "" if name not in job_map or ts > (job_map[name].get("_ts") or ""): job_map[name] = { "name": name, "type": s.get("jobType") or s.get("type") or "", "result": s.get("result") or "", "state": s.get("state") or "", "last_backup": ts, "schedule_enabled": True, "_ts": ts, } for j in job_map.values(): j.pop("_ts", None) jobs_data = list(job_map.values()) try: repos_data = await client.fetch_repositories() except Exception as e: errors.append(f"repos: {e}") try: repo_states = await client.fetch_repository_states() except Exception as e: errors.append(f"repo states: {e}") try: alarms_data = await client.fetch_alarms() except Exception as e: pass try: agents_data = await client.fetch_agents() except Exception as e: errors.append(f"agents: {e}") protection_groups = [] try: protection_groups = await client.fetch_protection_groups() except Exception as e: errors.append(f"protection groups: {e}") try: agent_policies = await client.fetch_agent_policies() except Exception as e: errors.append(f"agent policies: {e}") agent_policies = [] backups_data = [] try: backups_data = await client.fetch_backups() except Exception as e: errors.append(f"backups: {e}") restore_points = [] try: restore_points = await client.fetch_restore_points(200) except Exception as e: errors.append(f"restore points: {e}") for s in sessions_data: if isinstance(s.get("result"), dict): s["result"] = s["result"].get("result") or s["result"].get("Result") or "unknown" if "jobName" in s and "job_name" not in s: s["job_name"] = s["jobName"] if "creationTime" in s and "creation_time" not in s: s["creation_time"] = s["creationTime"] if "endTime" in s and "end_time" not in s: s["end_time"] = s["endTime"] job_last_backup = {} for s in sessions_data: s_name = s.get("jobName") or s.get("name") or s.get("Name") or "" s_end = s.get("endTime") or s.get("end_time") or s.get("creationTime") or "" if s_name and s_end: if s_name not in job_last_backup or s_end > job_last_backup[s_name]: job_last_backup[s_name] = s_end for j in jobs_data: if isinstance(j.get("result"), dict): j["result"] = j["result"].get("result") or j["result"].get("Result") or "unknown" if not j.get("result") and j.get("lastResult"): j["result"] = j["lastResult"] if not j.get("state") and j.get("status"): j["state"] = j["status"] if "scheduleEnabled" in j and "schedule_enabled" not in j: j["schedule_enabled"] = j["scheduleEnabled"] if "restorePointsToKeep" in j and "restore_points_to_keep" not in j: j["restore_points_to_keep"] = j["restorePointsToKeep"] if "nextRun" in j and "next_run" not in j: j["next_run"] = j["nextRun"] j_name = j.get("name") or j.get("Name") or "" if j_name in job_last_backup: j["last_backup"] = job_last_backup[j_name] rp_per_backup: dict[str, int] = {} for rp in restore_points: bid = rp.get("backupId") or "" if bid: rp_per_backup[bid] = rp_per_backup.get(bid, 0) + 1 backup_to_job: dict[str, str] = {} for b in backups_data: bid = b.get("id") or "" jid = b.get("jobId") or "" if bid and jid: backup_to_job[bid] = jid rp_per_job: dict[str, int] = {} for bid, count in rp_per_backup.items(): jid = backup_to_job.get(bid, "") if jid: rp_per_job[jid] = rp_per_job.get(jid, 0) + count for j in jobs_data: jid = j.get("id") or j.get("Id") or "" if jid in rp_per_job: j["actual_restore_points"] = rp_per_job[jid] job_count = len(jobs_data) success = sum(1 for j in jobs_data if _safe_str(j.get("result") or "").lower() in ("success", "true")) failed = sum(1 for j in jobs_data if _safe_str(j.get("result") or "").lower() in ("failed", "error", "false")) running = sum(1 for j in jobs_data if _safe_str(j.get("state") or "").lower() == "running") warning = sum(1 for j in jobs_data if "warning" in _safe_str(j.get("result") or "").lower()) disabled = sum(1 for j in jobs_data if j.get("schedule_enabled") is False or _safe_str(j.get("state") or "").lower() == "disabled") last_sessions = sessions_data[:20] if sessions_data else [] result = { "id": inst.id, "name": inst.name, "host": inst.host, "port": port, "version": client.version, "is_online": True, "user_role": role, "errors": errors, "jobs": jobs_data, "jobs_summary": { "total": job_count, "success": success, "failed": failed, "running": running, "warning": warning, "disabled": disabled, }, "disabled_policies": disabled, "recent_sessions": last_sessions, "repositories": repos_data, "repository_states": repo_states, "alarms": alarms_data, "agents": agents_data, "agent_policies": agent_policies, "protection_groups": protection_groups, "restore_points": restore_points[:50] if restore_points else [], } await set_cache(inst.id, "dashboard", result) return result except Exception as e: error_result = { "id": inst.id, "name": inst.name, "host": inst.host, "port": port, "version": ver, "is_online": False, "error": str(e), "jobs": [], "jobs_summary": {"total": 0, "success": 0, "failed": 0, "running": 0, "warning": 0, "disabled": 0}, "disabled_policies": 0, "recent_sessions": [], "repositories": [], "repository_states": [], "alarms": [], "agents": [], "agent_policies": [], "protection_groups": [], "restore_points": [], } return error_result finally: await client.close() async def _collect_all() -> list[dict]: instances = await get_instances() tasks = [_collect_instance(inst) for inst in instances if inst.is_active] if not tasks: return [] results = await asyncio.gather(*tasks, return_exceptions=True) output = [] for r in results: if isinstance(r, Exception): continue if r: output.append(r) return output def _build_summary(instance_data: list[dict]) -> dict: total_instances = len(instance_data) online = sum(1 for i in instance_data if i.get("is_online")) total_jobs = sum(i.get("jobs_summary", {}).get("total", 0) for i in instance_data) total_success = sum(i.get("jobs_summary", {}).get("success", 0) for i in instance_data) total_failed = sum(i.get("jobs_summary", {}).get("failed", 0) for i in instance_data) total_running = sum(i.get("jobs_summary", {}).get("running", 0) for i in instance_data) total_warning = sum(i.get("jobs_summary", {}).get("warning", 0) for i in instance_data) total_alarms = sum(len(i.get("alarms", [])) for i in instance_data) total_repos = sum(len(i.get("repositories", [])) for i in instance_data) return { "total_instances": total_instances, "online_instances": online, "offline_instances": total_instances - online, "total_jobs": total_jobs, "successful_jobs": total_success, "failed_jobs": total_failed, "running_jobs": total_running, "warning_jobs": total_warning, "total_alarms": total_alarms, "total_repositories": total_repos, } # ─── API Endpoints ───────────────────────────────────────────────── @app.get("/api/health") async def health(): return {"status": "ok"} # Instance CRUD @app.get("/api/instances", response_model=list[InstanceResponse]) async def list_instances(): instances = await get_instances() return [_serialize_instance(i) for i in instances] @app.post("/api/instances", response_model=InstanceResponse, status_code=201) async def create_instance(data: InstanceCreate): port = data.port or port_for_version(data.version) inst = await add_instance( name=data.name, host=data.host, port=port, username=data.username, password=data.password, version=data.version, notes=data.notes, ) return _serialize_instance(inst) @app.get("/api/instances/{instance_id}", response_model=InstanceResponse) async def read_instance(instance_id: int): inst = await get_instance(instance_id) if not inst: raise HTTPException(404, "Instance not found") return _serialize_instance(inst) @app.put("/api/instances/{instance_id}", response_model=InstanceResponse) async def edit_instance(instance_id: int, data: InstanceUpdate): updates = data.model_dump(exclude_unset=True) if not updates: raise HTTPException(400, "No fields to update") inst = await update_instance(instance_id, **updates) if not inst: raise HTTPException(404, "Instance not found") return _serialize_instance(inst) @app.delete("/api/instances/{instance_id}") async def remove_instance(instance_id: int): ok = await delete_instance(instance_id) if not ok: raise HTTPException(404, "Instance not found") _cache.pop(f"instance_{instance_id}", None) return {"ok": True} # Test connection — MUST be before {instance_id} routes to avoid path conflicts @app.post("/api/test-connection") async def test_connection_raw(data: InstanceCreate): port = data.port or port_for_version(data.version) ver = data.version if ver == "auto": ver = detect_version_from_host(data.host) port = port_for_version(ver) client = VeeamClient(data.host, port, data.username, data.password, ver) try: role = await client.connect() return {"ok": True, "role": role, "version": client.version} except Exception as e: return JSONResponse(status_code=502, content={"ok": False, "error": str(e)}) finally: await client.close() @app.post("/api/instances/{instance_id}/test") async def test_instance(instance_id: int): inst = await get_instance(instance_id) if not inst: raise HTTPException(404, "Instance not found") result = await _collect_instance(inst) if result and result.get("is_online"): return {"ok": True, "role": result.get("user_role"), "version": result.get("version")} return JSONResponse( status_code=502, content={"ok": False, "error": result.get("error", "Unknown error") if result else "No response"} ) # Dashboard @app.get("/api/dashboard") async def dashboard(force_refresh: bool = False): if not force_refresh: cached = _cached("dashboard_aggregate") if cached: return cached instance_data = await _collect_all() summary = _build_summary(instance_data) result = { "instances": instance_data, "summary": summary, "updated_at": datetime.utcnow().isoformat(), } _set_cache_inmem("dashboard_aggregate", result) return result # Per-instance data @app.get("/api/instances/{instance_id}/data") async def instance_data(instance_id: int, force_refresh: bool = False): inst = await get_instance(instance_id) if not inst: raise HTTPException(404, "Instance not found") if not force_refresh: cached = _cached(f"instance_{instance_id}") if cached: return cached result = await _collect_instance(inst) _set_cache_inmem(f"instance_{instance_id}", result) if not result.get("is_online"): return JSONResponse(status_code=502, content=result) return result # Serve frontend @app.get("/") async def index(): return FileResponse(str(BASE_DIR / "static" / "index.html")) if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host=settings.host, port=settings.port, reload=True)