fix: factorize models to dataclasss (easier output and default settings)

src/pvecontrol/models/backup_job.py

@@ -1,39 +1,44 @@
-class PVEBackupJob:
-    """Proxmox VE Backup Job"""
+from dataclasses import dataclass, field
+from typing import Optional
+
+from pvecontrol.models.storage import StorageShared
+
+
+@dataclass
+class PVEBackupJobData:
+    all: int = field(default=0)
+    compress: object = None
+    enabled: object = None
+    exclude: str = field(default="")
+    fleecing: object = None
+    mode: str = field(default="")
+    next_run: int = field(default=0)
+    node: str = field(default="")
+    notes_template: str = field(default="")
+    pool: str = field(default="")
+    prune_backups: object = (None,)
+    schedule: str = field(default="")
+    storage: Optional["StorageShared"] = None
+    type: str = field(default="")
+    vmid: str = field(default="")
 
-    _default_kwargs = {
-        "all": 0,
-        "compress": None,
-        "enabled": None,
-        "exclude": "",
-        "fleecing": None,
-        "mode": None,
-        "next-run": None,
-        "node": None,
-        "notes-template": None,
-        "pool": None,
-        "prune-backups": None,
-        "schedule": None,
-        "storage": None,
-        "type": None,
-        "vmid": "",
-    }
+
+class PVEBackupJob(PVEBackupJobData):
+    """Proxmox VE Backup Job"""
 
     def __init__(self, backup_id, **kwargs):
         self.id = backup_id
-
-        for k, v in self._default_kwargs.items():
-            self.__setattr__(k, kwargs.get(k, v))
+        _values = {k: v for k, v in kwargs.items() if hasattr(PVEBackupJobData, k)}
+        super().__init__(**_values)
 
         self.all = self.all == 1
-        self.vmid = self.vmid.split(",")
-        self.exclude = self.exclude.split(",")
+        if isinstance(self.vmid, str):
+            self.vmid = list(self.vmid.split(","))
+        if isinstance(self.exclude, str):
+            self.exclude = self.exclude.split(",")
 
     def __str__(self):
-        output = f"Vm(s): {self.vmid}\n" + f"Id: {self.id}\n"
-        for key in self._default_kwargs:
-            output += f"{key.capitalize()}: {self.__getattribute__(key)}\n"
-        return output
+        return "\n".join([f"{k.capitalize()}: {v}" for k, v in self.__dict__.items()])
 
     def is_selection_matching(self, vm):
         if self.node is not None and self.node != vm.node:

src/pvecontrol/models/cluster.py

@@ -29,28 +29,10 @@ def __init__(self, name, host, config, timeout, verify_ssl=False, **auth):
         self._ha = None
         self._backups = None
         self._backup_jobs = None
-        self._initstatus()
-
-    def _initstatus(self):
         self.status = self.api.cluster.status.get()
         self.resources = self.api.cluster.resources.get()
-
-        self.nodes = []
-        for node in self.resources_nodes:
-            self.nodes.append(
-                PVENode(
-                    self,
-                    node["node"],
-                    node["status"],
-                    kwargs=node,
-                )
-            )
-
-        self.storages = []
-        for storage in self.resources_storages:
-            self.storages.append(
-                PVEStorage(self.api, storage.pop("node"), storage.pop("id"), storage.pop("shared"), **storage)
-            )
+        self.nodes = [PVENode(self, **node) for node in self.resources_nodes]
+        self.storages = [PVEStorage(self.api, **storage) for storage in self.resources_storages]
 
     @staticmethod
     def create_from_config(cluster_name):
@@ -249,9 +231,7 @@ def backups(self):
                 logging.debug("Find storage: %s", (str(item)))
                 for backup in item["storage"].get_content("backup"):
                     logging.debug("New vm backup: %s", (str(backup)))
-                    self._backups.append(
-                        PVEVolume(backup.pop("volid"), backup.pop("format"), backup.pop("size"), **backup)
-                    )
+                    self._backups.append(PVEVolume(**backup))
         return self._backups
 
     @property

src/pvecontrol/models/node.py

@@ -1,4 +1,6 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Any, List, Optional
 
 from pvecontrol.utils import defaulter
 from pvecontrol.models.vm import PVEVm, VmStatus
@@ -13,24 +15,31 @@ class NodeStatus(Enum):
     OFFLINE = 2
 
 
-class PVENode:
-    """A proxmox VE Node"""
+@dataclass
+class PVENodeData:
+    node: str = field(default="")
+    status: Optional["NodeStatus"] = None
+    cluster: Any = None
+    cpu: int = field(default=0)
+    allocatedcpu: int = field(default=0)
+    maxcpu: int = field(default=0)
+    mem: int = field(default=0)
+    allocatedmem: int = field(default=0)
+    maxmem: int = field(default=0)
+    disk: int = field(default=0)
+    maxdisk: int = field(default=0)
+    vms: List = field(default_factory=list)
+
 
-    def __init__(self, cluster, node, status, kwargs=None):
-        if not kwargs:
-            kwargs = {}
+class PVENode(PVENodeData):
+    """A proxmox VE Node"""
 
-        self.node = node
-        self.status = NodeStatus[status.upper()]
+    def __init__(self, cluster, **kwargs):
+        _values = {k: v for k, v in kwargs.items() if hasattr(PVENodeData, k)}
+        super().__init__(**_values)
+        if isinstance(self.status, str):
+            self.status = NodeStatus[self.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)
         self._init_vms()
         self._init_allocatedmem()
         self._init_allocatedcpu()
@@ -49,7 +58,7 @@ def __str__(self):
     def _init_vms(self):
         self.vms = []
         if self.status == NodeStatus.ONLINE:
-            self.vms = [PVEVm(self.api, self.node, vm["vmid"], vm["status"], vm) for vm in self.resources_vms]
+            self.vms = [PVEVm(self.api, **vm) for vm in self.resources_vms]
 
     def _init_allocatedmem(self):
         """Compute the amount of memory allocated to running VMs"""

src/pvecontrol/models/storage.py

@@ -1,4 +1,6 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Optional
 
 from pvecontrol.models.volume import PVEVolume
 
@@ -21,30 +23,34 @@ class StorageShared(Enum):
     SHARED = 1
 
 
-class PVEStorage:
+@dataclass
+class PVEStorageData:
+    api: object = None
+    node: str = field(default="")
+    id: str = field(default="")
+    shared: Optional["StorageShared"] = None
+    type: str = field(default="")
+    storage: str = field(default="")
+    maxdisk: int = field(default=0)
+    disk: int = field(default=0)
+    plugintype: str = field(default="")
+    status: str = field(default="")
+    content: str = field(default="")
+
+
+class PVEStorage(PVEStorageData):
     """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.short_id = storage_id.split("/")[-1]
-        self.node = node
+    _api = None
+
+    def __init__(self, api, **kwargs):
+        super().__init__(**kwargs)
+        if isinstance(self.id, str):
+            self.short_id = self.id.rsplit("/", maxsplit=1)[-1]
         self._api = api
         self._content = {}
         self._details = {}
-
-        self.shared = STORAGE_SHARED_ENUM[shared]
-
-        for k, v in self._default_kwargs.items():
-            self.__setattr__(k, kwargs.get(k, v))
+        self.shared = StorageShared[STORAGE_SHARED_ENUM[self.shared].upper()]
 
     @property
     def details(self):
@@ -58,7 +64,7 @@ def get_grouped_list(proxmox):
         storages = {}
         for storage in proxmox.storages:
             value = {"storage": storage, "nodes": [], "usage": f"{storage.percentage:.1f}%"}
-            if StorageShared[storage.shared.upper()] == StorageShared.SHARED:
+            if storage.shared == StorageShared.SHARED:
                 storages[storage.storage] = storages.get(storage.storage, value)
                 storages[storage.storage]["nodes"] += [storage.node]
             else:
@@ -87,10 +93,7 @@ def percentage(self):
 
     @property
     def images(self):
-        images = []
-        for image in self.get_content("images"):
-            images.append(PVEVolume(image.pop("volid"), image.pop("format"), image.pop("size"), **image))
-        return images
+        return [PVEVolume(**image) for image in self.get_content("images")]
 
     def get_content(self, content_type=None):
         if content_type not in self._content:

src/pvecontrol/models/vm.py

@@ -1,4 +1,6 @@
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import Optional
 
 
 COLUMNS = ["vmid", "name", "status", "node", "cpus", "maxmem", "maxdisk", "tags"]
@@ -13,31 +15,37 @@ class VmStatus(Enum):
     PRELAUNCH = 5
 
 
-class PVEVm:
+@dataclass
+class PVEVmData:
+    api: object = None
+    node: str = field(default="")
+    status: Optional["VmStatus"] = None
+    vmid: int = field(default=0)
+    name: str = field(default="")
+    lock: str = field(default="")
+    pool: str = field(default="")
+    maxcpu: int = field(default=0)
+    maxdisk: int = field(default=0)
+    maxmem: int = field(default=0)
+    uptime: int = field(default=0)
+    tags: str = field(default="")
+    template: int = field(default=0)
+
+
+class PVEVm(PVEVmData):
     """Proxmox VE Qemu VM"""
 
     _api = None
+    _config = 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
+    def __init__(self, api, **kwargs):
+        _values = {k: v for k, v in kwargs.items() if hasattr(PVEVmData, k)}
+        super().__init__(**_values)
+        if isinstance(self.status, str):
+            self.status = VmStatus[self.status.upper()]
         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", "")
-
-        self._config = None
+        self.tags = set(filter(None, self.tags.split(";")))
+        self.cpus = self.maxcpu
 
     @property
     def config(self):

src/pvecontrol/models/volume.py

@@ -1,30 +1,35 @@
-class PVEVolume:
-    """Proxmox VE Volume"""
+from dataclasses import dataclass, field
+
+
+@dataclass
+class PVEVolumeData:
+    volid: str = field(default="")
+    volume_format: str = field(default="")
+    size: int = field(default=0)
+    content: str = field(default="")
 
-    _default_kwargs = {
-        "content": None,
-        "ctime": None,
-        "encrypted": None,
-        "notes": None,
-        "parent": None,
-        "path": None,
-        "protected": None,
-        "subtype": None,
-        "used": None,
-        "verification": None,
-        "vmid": None,
-    }
+    # is a string in API... convert it upon super() call...
+    ctime: int = field(default=0)
+    # unsure with those types... just porting _default_kwargs
+    encrypted: object = None
+    note: object = None
+    parent: object = None
+    path: object = None
+    protected: object = None
+    subtype: object = None
+    used: object = None
+    verification: object = None
+    vmid: int = field(default=0)
 
-    def __init__(self, volid, volume_format, size, **kwargs):
-        self.volid = volid
-        self.format = volume_format
-        self.size = size
 
-        for k, v in self._default_kwargs.items():
-            self.__setattr__(k, kwargs.get(k, v))
+class PVEVolume(PVEVolumeData):
+    """Proxmox VE Volume"""
+
+    def __init__(self, **kwargs):
+        _values = {k: v for k, v in kwargs.items() if hasattr(PVEVolumeData, k)}
+        super().__init__(**_values)
+        if isinstance(self.ctime, str):
+            self.ctime = int(self.ctime)
 
     def __str__(self):
-        output = f"Id: {self.volid}\n"
-        for key in self._default_kwargs:
-            output += f"{key.capitalize()}: {self.__getattribute__(key)}\n"
-        return output
+        return "\n".join([f"{k.capitalize()}: {v}" for k, v in self.__dict__.items()])