import json import sys import yaml from pathlib import Path from typing import Any, AsyncGenerator from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from app.auth import verify_token from app.ops_runner import ( run_command_host, stream_ops_host, new_op_id, is_cancelled, clear_cancelled, _BACKUP_TIMEOUT, ) sys.path.insert(0, "/opt/infrastructure") from toolkit.discovery import all_projects # noqa: E402 from toolkit.descriptor import find as find_project # noqa: E402 router = APIRouter() class ScheduleUpdate(BaseModel): enabled: bool = True schedule: str = "03:00" environments: list[str] | None = None command: str | None = None offsite: bool = False offsite_envs: list[str] | None = None retention_local_days: int | None = 7 retention_offsite_days: int | None = 30 @router.get("/", summary="Get backup schedules for all projects") async def get_schedules( _: str = Depends(verify_token), ) -> list[dict[str, Any]]: """Return backup schedule config for each project from descriptors.""" projects = all_projects() result = [] for name, desc in sorted(projects.items()): backup = desc.backup or {} result.append({ "project": name, "has_backup_dir": bool(backup.get("backup_dir") or backup.get("volumes")), "has_cli": desc.sync.get("type") == "cli", "static": desc.type == "static", "infrastructure": desc.type == "infrastructure", "environments": [e.name for e in desc.environments], # Backup schedule fields "enabled": backup.get("enabled", False), "schedule": backup.get("schedule", ""), "backup_environments": backup.get("environments"), "command": backup.get("command"), "offsite": backup.get("offsite", False), "offsite_envs": backup.get("offsite_envs"), "retention_local_days": backup.get("retention", {}).get("local_days"), "retention_offsite_days": backup.get("retention", {}).get("offsite_days"), }) return result @router.put("/{project}", summary="Update backup schedule for a project") async def update_schedule( project: str, body: ScheduleUpdate, _: str = Depends(verify_token), ) -> dict[str, Any]: """Update the backup schedule in project.yaml and regenerate timers.""" desc = find_project(project) if not desc: raise HTTPException(status_code=404, detail=f"Project '{project}' not found") # Read the full project.yaml yaml_path = Path(desc.path) / "project.yaml" try: with open(yaml_path) as f: project_yaml = yaml.safe_load(f) or {} except FileNotFoundError: raise HTTPException(status_code=404, detail=f"project.yaml not found at {yaml_path}") # Update only the backup block fields that were sent backup = project_yaml.get("backup", {}) backup["enabled"] = body.enabled backup["schedule"] = body.schedule if body.command: backup["command"] = body.command if body.environments: backup["environments"] = body.environments if body.offsite: backup["offsite"] = True if body.offsite_envs: backup["offsite_envs"] = body.offsite_envs else: backup["offsite"] = False retention = backup.get("retention", {}) if body.retention_local_days is not None: retention["local_days"] = body.retention_local_days if body.offsite and body.retention_offsite_days is not None: retention["offsite_days"] = body.retention_offsite_days if retention: backup["retention"] = retention project_yaml["backup"] = backup new_yaml = yaml.dump(project_yaml, default_flow_style=False, sort_keys=False) write_result = await run_command_host([ "bash", "-c", f"cat > {yaml_path} << 'YAMLEOF'\n{new_yaml}YAMLEOF" ]) if not write_result["success"]: raise HTTPException( status_code=500, detail=f"Failed to write project.yaml: {write_result['error']}" ) gen_result = await run_command_host([ "/usr/local/bin/ops", "gen-timers" ]) if not gen_result["success"]: raise HTTPException( status_code=500, detail=f"Failed to regenerate timers: {gen_result['error'] or gen_result['output']}" ) return { "success": True, "project": project, "backup": backup, "gen_timers_output": gen_result["output"], } def _sse(payload: dict) -> str: return f"data: {json.dumps(payload)}\n\n" def _now() -> str: return datetime.now(timezone.utc).isoformat() async def _run_now_stream(project: str) -> AsyncGenerator[str, None]: """Run backup for a project (all configured envs).""" op_id = new_op_id() yield _sse({"op_id": op_id}) desc = find_project(project) if not desc: yield _sse({"line": f"[error] Project '{project}' not found", "timestamp": _now()}) yield _sse({"done": True, "success": False}) return envs = [e.name for e in desc.environments] or [None] success = True for env in envs: if is_cancelled(op_id): yield _sse({"line": "Cancelled.", "timestamp": _now()}) yield _sse({"done": True, "success": False, "cancelled": True}) clear_cancelled(op_id) return cmd = ["backup", project] if env: cmd.append(env) label = f"{project}/{env}" if env else project yield _sse({"line": f"=== Backing up {label} ===", "timestamp": _now()}) async for line in stream_ops_host(cmd, timeout=_BACKUP_TIMEOUT, op_id=op_id): yield _sse({"line": line, "timestamp": _now()}) if line.startswith("[error]") or line.startswith("ERROR"): success = False if is_cancelled(op_id): yield _sse({"done": True, "success": False, "cancelled": True}) else: yield _sse({"done": True, "success": success}) clear_cancelled(op_id) @router.get("/{project}/run", summary="Run backup now (streaming)") async def run_backup_now( project: str, _: str = Depends(verify_token), ) -> StreamingResponse: """Trigger an immediate backup for a project, streaming output via SSE.""" desc = find_project(project) if not desc: raise HTTPException(status_code=404, detail=f"Project '{project}' not found") return StreamingResponse( _run_now_stream(project), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, )