diff --git a/functions.py b/functions.py index cba48f5..5036363 100644 --- a/functions.py +++ b/functions.py @@ -1,22 +1,23 @@ import logging -from typing import Optional, Tuple + +from proxmoxer import ProxmoxAPI logger = logging.getLogger(__name__) def create_vm( - proxmox, - node: Optional[str] = None, - vmid: Optional[int] = None, - name: Optional[str] = None, + proxmox: ProxmoxAPI, + node: str | None = None, + vmid: int | None = None, + name: str | None = None, prefix: str = 'stepchenkoas', cores: int = 2, memory: int = 2048, disk_size: int = 16, start: int = 1, -): +) -> None: if node is None: nodes = proxmox.nodes.get() - nodes = list(filter(lambda node: node['status'] == 'online', nodes)) + nodes = list(filter(lambda node: node['status'] == 'online', nodes)) # pyright: ignore node = min(nodes, key=lambda node: node['disk'])['node'] if vmid is None: @@ -47,10 +48,10 @@ def create_vm( def delete_vm( - proxmox, - node, - vmid, -): + proxmox: ProxmoxAPI, + node: str, + vmid: int, +) -> None: proxmox.nodes(node).qemu(vmid).delete( node=node, vmid=vmid, @@ -59,15 +60,15 @@ def delete_vm( def clone_template( - proxmox, + proxmox: ProxmoxAPI, node: str, vmid: int, - newid: Optional[int] = None, - name: Optional[str] = None, + newid: int | None = None, + name: str | None = None, prefix: str ='stepchenkoas', -) -> Tuple[int, str]: +) -> tuple[int, str]: if newid is None: - newid = int(proxmox.cluster.get('nextid')) + newid = int(proxmox.cluster.get('nextid')) # pyright: ignore if name is None: name = f'{prefix}-{newid}' @@ -85,7 +86,7 @@ def clone_template( pool='Virt_LAB', storage='rbd-storage', target=node, - ) + ) # pyright: ignore # TODO: check task status logger.debug("UPID '%s'", upid) logger.info('VM %s was cloned to %s successfully!', vmid, newid) @@ -93,11 +94,11 @@ def clone_template( return (newid, upid) def get_vm_ip( - proxmox, - node: 'str', + proxmox: ProxmoxAPI, + node: str, vmid: int, -): - res = proxmox.nodes(node).qemu(vmid).agent('network-get-interfaces').get() +) -> str: + res: dict = proxmox.nodes(node).qemu(vmid).agent('network-get-interfaces').get() # pyright: ignore vm_ip_address = None for net in res['result']: if net['name'] == 'eth0' or net['name'] == 'ens19': diff --git a/main.py b/main.py index 450e8d3..e60d06b 100644 --- a/main.py +++ b/main.py @@ -1,7 +1,7 @@ import os import logging import urllib3 -from typing import Callable, List +from typing import Callable from time import sleep from pathlib import Path @@ -37,12 +37,13 @@ def main(): PROXMOX_HOST, user=PROXMOX_USER_FULL, password=PROXMOX_PASSWORD, verify_ssl=False ) - nodes = proxmox.nodes.get() + nodes: list[dict[str, str | int]] = proxmox.nodes.get() # pyright: ignore nodes = list(filter(lambda node: node['status'] == 'online', nodes)) - node = min(nodes, key=lambda node: node['disk'])['node'] - template_id = 374 - prefix = PROXMOX_USER + '-test-k8s' - vm_names = { + # node = min(nodes, key=lambda node: node['disk'])['node'] + node: str = 'pve05' + template_id: int = 374 + prefix: str = PROXMOX_USER + '-test-k8s' + vm_names: dict[str, str] = { 'master': f'{prefix}-master', 'worker1': f'{prefix}-node1', 'worker2': f'{prefix}-node2', @@ -121,12 +122,12 @@ def main(): sleep(40) logger.info('VMs are running!') - vm_ip_master = get_vm_ip(proxmox, node, vm_ids['master']) - vm_ip_worker1 = get_vm_ip(proxmox, node, vm_ids['worker1']) - vm_ip_worker2 = get_vm_ip(proxmox, node, vm_ids['worker2']) + vm_ip_master: str = get_vm_ip(proxmox, node, vm_ids['master']) + vm_ip_worker1: str = get_vm_ip(proxmox, node, vm_ids['worker1']) + vm_ip_worker2: str = get_vm_ip(proxmox, node, vm_ids['worker2']) proxmox.logout() - tmp_path = './tmp' + tmp_path: str = './tmp' if not os.path.exists(tmp_path): os.makedirs(tmp_path)