1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
| | from typing import Any
| |
| | from fastapi import APIRouter, Depends, HTTPException
| |
| | from app.auth import verify_token
| | from app.ops_runner import run_ops, run_ops_json, run_ops_host, _BACKUP_TIMEOUT
| |
| | router = APIRouter()
| |
| |
| | @router.get("/", summary="List local backups")
| | async def list_backups(
| | _: str = Depends(verify_token),
| | ) -> list[dict[str, Any]]:
| | """Returns a list of local backup records from `ops backups --json`."""
| | result = await run_ops_json(["backups"])
| | if not result["success"]:
| | raise HTTPException(status_code=500, detail=f"Failed to list backups: {result['error']}")
| |
| | data = result["data"]
| | if isinstance(data, list):
| | return data
| | if isinstance(data, dict):
| | for key in ("backups", "data", "items"):
| | if key in data and isinstance(data[key], list):
| | return data[key]
| | return [data]
| | return []
| |
| |
| | @router.get("/offsite", summary="List offsite backups")
| | async def list_offsite_backups(
| | _: str = Depends(verify_token),
| | ) -> list[dict[str, Any]]:
| | """Returns a list of offsite backup records."""
| | all_backups = []
| | for project in ["mdf", "seriousletter"]:
| | result = await run_ops_json(["offsite", "list", project])
| | if result["success"] and isinstance(result["data"], list):
| | for b in result["data"]:
| | b["project"] = project
| | all_backups.extend(result["data"])
| | return all_backups
| |
| |
| | @router.post("/{project}/{env}", summary="Create a local backup")
| | async def create_backup(
| | project: str,
| | env: str,
| | _: str = Depends(verify_token),
| | ) -> dict[str, Any]:
| | """
| | Runs `ops backup {project} {env}` on the host.
| |
| | Runs via nsenter because ops backup delegates to project CLIs
| | that use host Python venvs.
| | """
| | result = await run_ops_host(["backup", project, env], timeout=_BACKUP_TIMEOUT)
| | if not result["success"]:
| | raise HTTPException(
| | status_code=500,
| | detail=f"Backup failed: {result['error'] or result['output']}",
| | )
| | return {
| | "success": True,
| | "output": result["output"],
| | "project": project,
| | "env": env,
| | }
| |
| |
| | @router.post("/offsite/upload/{project}/{env}", summary="Upload backup to offsite")
| | async def upload_offsite(
| | project: str,
| | env: str,
| | _: str = Depends(verify_token),
| | ) -> dict[str, Any]:
| | """Runs `ops offsite upload {project} {env}` on the host."""
| | result = await run_ops_host(
| | ["offsite", "upload", project, env], timeout=_BACKUP_TIMEOUT
| | )
| | if not result["success"]:
| | raise HTTPException(
| | status_code=500,
| | detail=f"Offsite upload failed: {result['error'] or result['output']}",
| | )
| | return {"success": True, "output": result["output"], "project": project, "env": env}
| |
| |
| | @router.post("/offsite/retention", summary="Apply offsite retention policy")
| | async def apply_retention(
| | _: str = Depends(verify_token),
| | ) -> dict[str, Any]:
| | """Runs `ops offsite retention` on the host."""
| | result = await run_ops_host(["offsite", "retention"], timeout=_BACKUP_TIMEOUT)
| | if not result["success"]:
| | raise HTTPException(
| | status_code=500,
| | detail=f"Retention policy failed: {result['error'] or result['output']}",
| | )
| | return {"success": True, "output": result["output"]}
|
|
