refactor: move backups and backups_job listing in PVECluster and PVEVm

src/pvecontrol/cluster.py

@@ -14,6 +14,8 @@ def __init__(self, name, host, user, password, config, timeout, verify_ssl=False
         self.api = ProxmoxAPI(host, user=user, password=password, timeout=timeout, verify_ssl=verify_ssl)
         self.name = name
         self.config = config
+        self._backups = None
+        self._backup_jobs = None
         self._initstatus()
 
     def _initstatus(self):
@@ -154,3 +156,17 @@ def metrics(self):
             "memory": self.memory_metrics(),
             "disk": self.disk_metrics(),
         }
+
+    @property
+    def backups(self):
+        if self._backups is None:
+            self._backups = []
+            for item in PVEStorage.get_grouped_list(self):
+                self._backups.extend(item["storage"].get_content("backup"))
+        return self._backups
+
+    @property
+    def backup_jobs(self):
+        if self._backup_jobs is None:
+            self._backup_jobs = self.api.cluster.backup.get()
+        return self._backup_jobs

src/pvecontrol/sanitycheck/tests/vm_backups.py

@@ -1,6 +1,5 @@
 from datetime import datetime, timedelta
 from pvecontrol.sanitycheck.checks import Check, CheckCode, CheckType, CheckMessage
-from pvecontrol.storage import PVEStorage
 
 
 class VmBackups(Check):
@@ -14,10 +13,9 @@ def run(self):
         self._check_backup_ran_recently(backuped_vms)
 
     def _check_is_backed_up(self):
-        backup_jobs = self.proxmox.api.cluster.backup.get()
         backuped_vms = []
         for vm in self.proxmox.vms():
-            vm_backup_jobs = self._get_vm_backup_jobs(vm, backup_jobs)
+            vm_backup_jobs = vm.get_backup_jobs(self.proxmox)
             vm_enabled_backup_ids = [backup["id"] for backup in vm_backup_jobs if backup["enabled"] == 1]
             if len(vm_enabled_backup_ids) > 0:
                 msg = f"Vm {vm.vmid} ({vm.name}) is associated to {len(vm_enabled_backup_ids)} enabled backup job(s)"
@@ -34,7 +32,7 @@ def _check_backup_ran_recently(self, vms):
         time_ago = f"{hm_ago[0]:02d} hour(s) and {hm_ago[1]:02d} minute(s) ago"
 
         for vm in vms:
-            last_backup = self._get_vm_last_backup(vm)
+            last_backup = vm.get_last_backup(self.proxmox)
             last_backup_time = datetime.fromtimestamp(last_backup["ctime"])
             msg_template = f"Vm {vm.vmid} ({vm.name}) has been backed up {{}} than {time_ago} ({last_backup_time.strftime('%Y-%m-%d %H:%M:%S')})"
             if last_backup_time > datetime.now() - timedelta(minutes=minutes_ago):
@@ -43,21 +41,3 @@ def _check_backup_ran_recently(self, vms):
             else:
                 msg = msg_template.format("more")
                 self.add_messages(CheckMessage(CheckCode.WARN, msg))
-
-    def _get_vm_backup_jobs(self, vm, backup_jobs):
-        vm_backups = []
-        for backup in backup_jobs:
-            if str(vm.vmid) in backup["vmid"].split(","):
-                vm_backups.append(backup)
-        return vm_backups
-
-    def _get_vm_backups(self, vm):
-        backups = []
-        for item in PVEStorage.get_grouped_list(self.proxmox):
-            backups.extend(item["storage"].get_content("backup"))
-
-        return [backup for backup in backups if backup["vmid"] == vm.vmid]
-
-    def _get_vm_last_backup(self, vm):
-        vm_backups = sorted(self._get_vm_backups(vm), key=lambda x: x["ctime"])
-        return vm_backups[-1] if len(vm_backups) > 0 else None

src/pvecontrol/vm.py

@@ -64,3 +64,17 @@ def migrate(self, target, online=False):
 
         upid = self._api.nodes(self.node).qemu(self.vmid).migrate.post(**options)
         return upid
+
+    def get_backup_jobs(self, proxmox):
+        vm_backups = []
+        for backup in proxmox.backup_jobs:
+            if str(self.vmid) in backup["vmid"].split(","):
+                vm_backups.append(backup)
+        return vm_backups
+
+    def get_backups(self, proxmox):
+        return [backup for backup in proxmox.backups if backup["vmid"] == self.vmid]
+
+    def get_last_backup(self, proxmox):
+        backups = sorted(self.get_backups(proxmox), key=lambda x: x["ctime"])
+        return backups[-1] if len(backups) > 0 else None