From cdd005d217fbb15ea81cb004acfae64a188ffb97 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Fri, 25 Apr 2025 17:03:58 +0200 Subject: [PATCH 1/4] fix: factorize models to dataclasss (easier output and default settings) --- src/pvecontrol/models/backup_job.py | 61 ++++++++++++---------- src/pvecontrol/models/cluster.py | 28 ++-------- src/pvecontrol/models/node.py | 41 +++++++++------ src/pvecontrol/models/storage.py | 51 +++++++++--------- src/pvecontrol/models/vm.py | 48 ++++++++++------- src/pvecontrol/models/volume.py | 55 ++++++++++--------- src/pvecontrol/sanitycheck/tests/ha_vms.py | 6 +-- src/tests/test_backup_job.py | 8 +-- src/tests/test_utils.py | 6 +-- 9 files changed, 158 insertions(+), 146 deletions(-) diff --git a/src/pvecontrol/models/backup_job.py b/src/pvecontrol/models/backup_job.py index 71d0ecef..f230ca34 100644 --- a/src/pvecontrol/models/backup_job.py +++ b/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: diff --git a/src/pvecontrol/models/cluster.py b/src/pvecontrol/models/cluster.py index 150733ef..2fb51744 100644 --- a/src/pvecontrol/models/cluster.py +++ b/src/pvecontrol/models/cluster.py @@ -29,28 +29,12 @@ 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] + for bu in self.backup_jobs: + print(bu) @staticmethod def create_from_config(cluster_name): @@ -249,9 +233,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 diff --git a/src/pvecontrol/models/node.py b/src/pvecontrol/models/node.py index 2b37823f..4175e1eb 100644 --- a/src/pvecontrol/models/node.py +++ b/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""" diff --git a/src/pvecontrol/models/storage.py b/src/pvecontrol/models/storage.py index a4e4e952..96e887a5 100644 --- a/src/pvecontrol/models/storage.py +++ b/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: diff --git a/src/pvecontrol/models/vm.py b/src/pvecontrol/models/vm.py index 74d692be..c53641e2 100644 --- a/src/pvecontrol/models/vm.py +++ b/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): diff --git a/src/pvecontrol/models/volume.py b/src/pvecontrol/models/volume.py index df911ce9..3af808a3 100644 --- a/src/pvecontrol/models/volume.py +++ b/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()]) diff --git a/src/pvecontrol/sanitycheck/tests/ha_vms.py b/src/pvecontrol/sanitycheck/tests/ha_vms.py index fc78a6aa..881f607c 100644 --- a/src/pvecontrol/sanitycheck/tests/ha_vms.py +++ b/src/pvecontrol/sanitycheck/tests/ha_vms.py @@ -1,6 +1,6 @@ import re -from pvecontrol.models.storage import StorageShared +from pvecontrol.models.storage import STORAGE_SHARED_ENUM, PVEStorage, StorageShared from pvecontrol.sanitycheck.checks import Check, CheckType, CheckMessage, CheckCode @@ -36,8 +36,8 @@ def _check_disk_ha_consistency(self, ha_vms): if not isinstance(v, str): continue if regex_result := re.search(regex, v): - storage = self.proxmox.get_storage(regex_result.group(1)) - if storage is not None and StorageShared[storage.shared.upper()] != StorageShared.SHARED: + storage: PVEStorage = self.proxmox.get_storage(regex_result.group(1)) + if storage is not None and storage.shared != StorageShared.SHARED: result["disks"].append(k) if result["disks"]: vms_not_consistent.append(result) diff --git a/src/tests/test_backup_job.py b/src/tests/test_backup_job.py index 3cd497cf..b251827e 100644 --- a/src/tests/test_backup_job.py +++ b/src/tests/test_backup_job.py @@ -4,10 +4,10 @@ def test_is_selection_matching(): vms = [ - PVEVm(None, "node-0", 0, "running", {"pool": "pool-A"}), - PVEVm(None, "node-1", 1, "running", {"pool": "pool-A"}), - PVEVm(None, "node-0", 2, "running", {"pool": "pool-B"}), - PVEVm(None, "node-1", 3, "running", {"pool": "pool-B"}), + PVEVm(api=None, node="node-0", vmid=0, status="running", pool="pool-A"), + PVEVm(api=None, node="node-1", vmid=1, status="running", pool="pool-A"), + PVEVm(api=None, node="node-0", vmid=2, status="running", pool="pool-B"), + PVEVm(api=None, node="node-1", vmid=3, status="running", pool="pool-B"), ] def check_is_selection_matching_array(truth_table, backup_job): diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py index 83cbb21e..bd85f677 100644 --- a/src/tests/test_utils.py +++ b/src/tests/test_utils.py @@ -13,9 +13,9 @@ def test_render_output(): api = Mock() vms = [ - PVEVm(api, "pve-node-1", 100, "running"), - PVEVm(api, "pve-node-1", 101, "running"), - PVEVm(api, "pve-node-2", 102, "stopped"), + PVEVm(api=api, node="pve-node-1", vmid=100, status="running"), + PVEVm(api=api, node="pve-node-2", vmid=101, status="running"), + PVEVm(api=api, node="pve-node-3", vmid=102, status="running"), ] output_text = render_output(vms, columns=COLUMNS, output=OutputFormats.TEXT) From 792846b185ca02692c9045dd9e9d1972c6acad5a Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Fri, 25 Apr 2025 17:06:08 +0200 Subject: [PATCH 2/4] blackify new dataclass... --- src/pvecontrol/models/backup_job.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pvecontrol/models/backup_job.py b/src/pvecontrol/models/backup_job.py index f230ca34..04b9ab8b 100644 --- a/src/pvecontrol/models/backup_job.py +++ b/src/pvecontrol/models/backup_job.py @@ -16,11 +16,11 @@ class PVEBackupJobData: node: str = field(default="") notes_template: str = field(default="") pool: str = field(default="") - prune_backups: object = None, + prune_backups: object = (None,) schedule: str = field(default="") storage: Optional["StorageShared"] = None type: str = field(default="") - vmid: str = field(default=""), + vmid: str = (field(default=""),) class PVEBackupJob(PVEBackupJobData): From 1b61134f288e67f8631b0384bfae0197c079b660 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Mon, 28 Apr 2025 16:24:51 +0200 Subject: [PATCH 3/4] satisfied pylint... --- src/pvecontrol/models/backup_job.py | 2 +- src/pvecontrol/sanitycheck/tests/ha_vms.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pvecontrol/models/backup_job.py b/src/pvecontrol/models/backup_job.py index 04b9ab8b..9601db3e 100644 --- a/src/pvecontrol/models/backup_job.py +++ b/src/pvecontrol/models/backup_job.py @@ -20,7 +20,7 @@ class PVEBackupJobData: schedule: str = field(default="") storage: Optional["StorageShared"] = None type: str = field(default="") - vmid: str = (field(default=""),) + vmid: str = field(default="") class PVEBackupJob(PVEBackupJobData): diff --git a/src/pvecontrol/sanitycheck/tests/ha_vms.py b/src/pvecontrol/sanitycheck/tests/ha_vms.py index 881f607c..1fcb61be 100644 --- a/src/pvecontrol/sanitycheck/tests/ha_vms.py +++ b/src/pvecontrol/sanitycheck/tests/ha_vms.py @@ -1,6 +1,6 @@ import re -from pvecontrol.models.storage import STORAGE_SHARED_ENUM, PVEStorage, StorageShared +from pvecontrol.models.storage import PVEStorage, StorageShared from pvecontrol.sanitycheck.checks import Check, CheckType, CheckMessage, CheckCode From 41edf5d8cfadf0e2a6cc86dd0900c93972f62680 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Wed, 28 May 2025 14:45:40 +0200 Subject: [PATCH 4/4] remove useless print... --- src/pvecontrol/models/cluster.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/pvecontrol/models/cluster.py b/src/pvecontrol/models/cluster.py index 2fb51744..b5ece140 100644 --- a/src/pvecontrol/models/cluster.py +++ b/src/pvecontrol/models/cluster.py @@ -33,8 +33,6 @@ def __init__(self, name, host, config, timeout, verify_ssl=False, **auth): self.resources = self.api.cluster.resources.get() self.nodes = [PVENode(self, **node) for node in self.resources_nodes] self.storages = [PVEStorage(self.api, **storage) for storage in self.resources_storages] - for bu in self.backup_jobs: - print(bu) @staticmethod def create_from_config(cluster_name):