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