diff --git a/src/pvecontrol/models/node.py b/src/pvecontrol/models/node.py index 2b37823..725d8b1 100644 --- a/src/pvecontrol/models/node.py +++ b/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() diff --git a/src/pvecontrol/models/storage.py b/src/pvecontrol/models/storage.py index a20d7aa..f9ce1b1 100644 --- a/src/pvecontrol/models/storage.py +++ b/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 diff --git a/src/pvecontrol/models/task.py b/src/pvecontrol/models/task.py index 51299bc..43653e5 100644 --- a/src/pvecontrol/models/task.py +++ b/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: diff --git a/src/pvecontrol/models/vm.py b/src/pvecontrol/models/vm.py index 74d692b..2c79bda 100644 --- a/src/pvecontrol/models/vm.py +++ b/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] diff --git a/src/pvecontrol/utils.py b/src/pvecontrol/utils.py index 0ca2b8d..d40cd86 100644 --- a/src/pvecontrol/utils.py +++ b/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