fix: switch main PVE* to dataclasses, harmonize defaults

src/pvecontrol/models/node.py

@@ -1,6 +1,8 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Any, Dict, List
 
-from pvecontrol.utils import defaulter
+from pvecontrol.utils import defaulter, dict_to_attr
 from pvecontrol.models.vm import PVEVm, VmStatus
 
 
@@ -13,24 +15,28 @@ class NodeStatus(Enum):
     OFFLINE = 2
 
 
+@dataclass
 class PVENode:
     """A proxmox VE Node"""
 
-    def __init__(self, cluster, node, status, kwargs=None):
-        if not kwargs:
-            kwargs = {}
-
-        self.node = node
-        self.status = NodeStatus[status.upper()]
-        self.cluster = cluster
-        self.cpu = kwargs.get("cpu", 0)
-        self.allocatedcpu = 0
-        self.maxcpu = kwargs.get("maxcpu", 0)
-        self.mem = kwargs.get("mem", 0)
-        self.allocatedmem = 0
-        self.maxmem = kwargs.get("maxmem", 0)
-        self.disk = kwargs.get("disk", 0)
-        self.maxdisk = kwargs.get("maxdisk", 0)
+    cluster: object
+    node: str
+    status: NodeStatus
+    kwargs: Dict[str, Any] = field(default_factory=dict)
+
+    cpu: int = field(init=True, default=0)
+    allocatedcpu: int = field(init=True, default=0)
+    maxcpu: int = field(init=True, default=0)
+    mem: int = field(init=True, default=0)
+    allocatedmem: int = field(init=True, default=0)
+    maxmem: int = field(init=True, default=0)
+    disk: int = field(init=True, default=0)
+    maxdisk: int = field(init=True, default=0)
+    vms: List[PVEVm] = field(default_factory=list)
+
+    def __post_init__(self):
+        dict_to_attr(self, "kwargs")
+        self.status = NodeStatus[self.status.upper()]
         self._init_vms()
         self._init_allocatedmem()
         self._init_allocatedcpu()

src/pvecontrol/models/storage.py

@@ -1,4 +1,9 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Any, Dict
+
+from proxmoxer import ProxmoxAPI
+from pvecontrol.utils import dict_to_attr
 
 
 STORAGE_SHARED_ENUM = ["local", "shared"]
@@ -19,28 +24,29 @@ class StorageShared(Enum):
     SHARED = 1
 
 
+@dataclass
 class PVEStorage:
     """Proxmox VE Storage"""
 
-    _default_kwargs = {
-        "storage": None,
-        "maxdisk": None,
-        "disk": None,
-        "plugintype": None,
-        "status": None,
-        "test": None,
-    }
-
-    def __init__(self, api, node, storage_id, shared, **kwargs):
-        self.id = storage_id
-        self.node = node
-        self._api = api
-        self._content = {}
+    api: ProxmoxAPI
+    node: str
+    storage_id: str
+    shared: StorageShared
+    kwargs: Dict[str, Any] = field(default_factory=dict)
 
-        self.shared = STORAGE_SHARED_ENUM[shared]
+    storage: Any = field(init=True, default=None)
+    maxdisk: int = field(init=True, default=0)
+    disk: int = field(init=True, default=0)
+    plugintype: str = field(init=True, default="")
+    status: str = field(init=True, default="")
+    test: str = field(init=True, default="")
+    content: dict = field(default_factory=dict)
+    type: str = field(init=True, default="")
 
-        for k, v in self._default_kwargs.items():
-            self.__setattr__(k, kwargs.get(k, v))
+    def __post_init__(self):
+        dict_to_attr(self, "kwargs")
+        self.shared = STORAGE_SHARED_ENUM[self.shared]
+        self._content = {}
 
     @staticmethod
     def get_grouped_list(proxmox):
@@ -51,8 +57,8 @@ def get_grouped_list(proxmox):
                 storages[storage.storage] = storages.get(storage.storage, value)
                 storages[storage.storage]["nodes"] += [storage.node]
             else:
-                storages[storage.id] = value
-                storages[storage.id]["nodes"] += [storage.node]
+                storages[storage.storage_id] = value
+                storages[storage.storage_id]["nodes"] += [storage.node]
 
         return storages.values()
 
@@ -62,12 +68,12 @@ def percentage(self):
 
     def get_content(self, content_type=None):
         if content_type not in self._content:
-            short_id = self.id.split("/")[-1]
-            self._content[content_type] = self._api.nodes(self.node).storage(short_id).content.get(content=content_type)
+            short_id = self.storage_id.split("/")[-1]
+            self._content[content_type] = self.api.nodes(self.node).storage(short_id).content.get(content=content_type)
         return self._content[content_type]
 
     def __str__(self):
-        output = f"Node: {self.node}\n" + f"Id: {self.id}\n"
+        output = f"Node: {self.node}\n" + f"Id: {self.storage_id}\n"
         for key in self._default_kwargs:
             output += f"{key.capitalize()}: {self.__getattribute__(key)}\n"
         return output

src/pvecontrol/models/task.py

@@ -1,6 +1,8 @@
+from dataclasses import dataclass
 from enum import Enum
 import proxmoxer.core
 from proxmoxer.tools import Tasks
+from proxmoxer import ProxmoxAPI
 
 
 COLUMNS = [
@@ -20,16 +22,16 @@ class TaskRunningStatus(Enum):
     VANISHED = 2
 
 
+@dataclass
 class PVETask:
     """Proxmox VE Task"""
 
-    _api = None
+    api: ProxmoxAPI
+    upid: str
 
-    def __init__(self, api, upid):
-        task = Tasks.decode_upid(upid)
+    def __post_init__(self):
+        task = Tasks.decode_upid(self.upid)
 
-        self._api = api
-        self.upid = upid
         self.node = task["node"]
         self.starttime = task["starttime"]
         self.type = task["type"]
@@ -41,7 +43,7 @@ def __init__(self, api, upid):
         self.refresh()
 
     def log(self, limit=0, start=0):
-        return self._api.nodes(self.node).tasks(self.upid).log.get(limit=limit, start=start)
+        return self.api.nodes(self.node).tasks(self.upid).log.get(limit=limit, start=start)
 
     def running(self):
         return self.runningstatus == TaskRunningStatus.RUNNING
@@ -54,7 +56,7 @@ def refresh(self):
         #    if self.node != NodeStatus.online:
         #      return
         try:
-            status = self._api.nodes(self.node).tasks(self.upid).status.get()
+            status = self.api.nodes(self.node).tasks(self.upid).status.get()
         # Some task information can be vanished over time (tasks status files removed from the node filesystem)
         # In this case API return an error and we consider this tasks vanished and don't get more informations
         except proxmoxer.core.ResourceException:

src/pvecontrol/models/vm.py

@@ -1,4 +1,8 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Any, Dict
+from proxmoxer import ProxmoxAPI
+from pvecontrol.utils import dict_to_attr
 
 
 COLUMNS = ["vmid", "name", "status", "node", "cpus", "maxmem", "maxdisk", "tags"]
@@ -13,36 +17,36 @@ class VmStatus(Enum):
     PRELAUNCH = 5
 
 
+@dataclass
 class PVEVm:
     """Proxmox VE Qemu VM"""
 
-    _api = None
-
-    def __init__(self, api, node, vmid, status, kwargs=None):
-        if not kwargs:
-            kwargs = {}
-
-        self.vmid = vmid
-        self.status = VmStatus[status.upper()]
-        self.node = node
-        self._api = api
-
-        self.name = kwargs.get("name", "")
-        self.lock = kwargs.get("lock", "")
-        self.cpus = kwargs.get("maxcpu", 0)
-        self.maxdisk = kwargs.get("maxdisk", 0)
-        self.maxmem = kwargs.get("maxmem", 0)
-        self.uptime = kwargs.get("uptime", 0)
-        self.tags = set(filter(None, kwargs.get("tags", "").split(";")))
-        self.template = kwargs.get("template", 0)
-        self.pool = kwargs.get("pool", "")
-
+    api: ProxmoxAPI
+    node: str
+    vmid: int
+    status: VmStatus
+    kwargs: Dict[str, Any] = field(default_factory=dict)
+
+    name: str = field(init=True, default="")
+    lock: str = field(init=True, default="")
+    cpus: int = field(init=True, default=0)
+    maxdisk: int = field(init=True, default=0)
+    maxmem: int = field(init=True, default=0)
+    uptime: int = field(init=True, default=0)
+    template: int = field(init=True, default=0)
+    pool: str = field(init=True, default="")
+    tags: str = field(init=True, default="")
+
+    def __post_init__(self):
+        dict_to_attr(self, "kwargs")
+        self.status = VmStatus[self.status.upper()]
+        self.tags = set(self.tags.split(";"))
         self._config = None
 
     @property
     def config(self):
         if not self._config:
-            self._config = self._api.nodes(self.node).qemu(self.vmid).config.get()
+            self._config = self.api.nodes(self.node).qemu(self.vmid).config.get()
 
         return self._config
 
@@ -59,30 +63,23 @@ def __str__(self):
             "tags",
             "template",
         ]
-        output = []
-        for k in str_keys:
-            output.append(f"{k}: {getattr(self, k)}")
+        output = [f"{k}: {getattr(self, k)}" for k in str_keys]
         return ", ".join(output)
 
     def migrate(self, target, online=False):
         options = {}
         options["node"] = self.node
         options["target"] = target
-        check = self._api.nodes(self.node).qemu(self.vmid).migrate.get(**options)
+        check = self.api.nodes(self.node).qemu(self.vmid).migrate.get(**options)
         #    logging.debug("Migration check: %s"%check)
         options["online"] = int(online)
         if len(check["local_disks"]) > 0:
             options["with-local-disks"] = int(True)
 
-        upid = self._api.nodes(self.node).qemu(self.vmid).migrate.post(**options)
-        return upid
+        return self.api.nodes(self.node).qemu(self.vmid).migrate.post(**options)
 
     def get_backup_jobs(self, proxmox):
-        vm_backup_jobs = []
-        for backup_job in proxmox.backup_jobs:
-            if backup_job.is_selection_matching(self):
-                vm_backup_jobs.append(backup_job)
-        return vm_backup_jobs
+        return [job for job in proxmox.backup_jobs if job.is_selection_matching(self)]
 
     def get_backups(self, proxmox):
         return [backup for backup in proxmox.backups if backup.vmid == self.vmid]

src/pvecontrol/utils.py

@@ -197,3 +197,10 @@ def defaulter(resource: dict, keys, default):
         if k not in resource.keys():
             resource[k] = default
     return resource
+
+
+def dict_to_attr(obj: object, base_attr: str):
+    kw = getattr(obj, base_attr, {}) or {}
+    for k, v in kw.items():
+        setattr(obj, k, v)
+    return obj