Refactor backend: Split models and proxmox_client into modules, add alembic
Some checks failed
Backend Tests / pytest (Python 3.12) (push) Failing after 2m27s

This commit is contained in:
2026-06-13 03:53:27 +05:00
parent 8a879f36fa
commit bfd92102cf
54 changed files with 6026 additions and 5137 deletions

View File

@@ -0,0 +1,143 @@
from typing import List, Dict, Optional, Union, Any
import time
import urllib3
from loguru import logger
class NetworkMixin:
def get_sdn_vnets(self) -> List[Dict]:
"""
Get all SDN VNets (Virtual Networks).
Returns:
List of VNets with vnet name, zone, alias, etc.
"""
if not self.proxmox:
return []
try:
vnets = self.proxmox.cluster.sdn.vnets.get()
return vnets if isinstance(vnets, list) else []
except Exception as e:
logger.error(f"Error getting SDN vnets: {e}")
return []
def get_sdn_zones(self) -> List[Dict]:
"""
Get all SDN zones.
Returns:
List of SDN zones with type, zone name, pending status
"""
if not self.proxmox:
return []
try:
zones = self.proxmox.cluster.sdn.zones.get()
return zones if isinstance(zones, list) else []
except Exception as e:
logger.error(f"Error getting SDN zones: {e}")
return []
def create_sdn_zone(self, zone: str, zone_type: str = "simple", **kwargs) -> Dict:
"""
Create a new SDN zone.
Args:
zone: Zone name (alphanumeric, max 8 chars)
zone_type: Type of zone (simple, vlan, qinq, vxlan, evpn)
**kwargs: Additional zone options (mtu, dns, reversedns, etc.)
Returns:
Result dict with success status
"""
if not self.proxmox:
return {"success": False, "error": "Not connected"}
try:
params = {
"zone": zone,
"type": zone_type,
}
params.update(kwargs)
self.proxmox.cluster.sdn.zones.post(**params)
logger.info(f"Created SDN zone: {zone} (type={zone_type})")
return {"success": True, "zone": zone}
except Exception as e:
logger.error(f"Error creating SDN zone {zone}: {e}")
return {"success": False, "error": str(e)}
def create_sdn_vnet(self, vnet: str, zone: str, tag: int = None,
alias: str = None, vlanaware: bool = False) -> Dict:
"""
Create a new SDN VNet.
Args:
vnet: VNet name (alphanumeric, max 8 chars)
zone: Zone where to create the VNet
tag: VLAN/VNI tag (optional)
alias: Human-readable alias
vlanaware: Enable VLAN-aware bridge
Returns:
Result dict with success status
"""
if not self.proxmox:
return {"success": False, "error": "Not connected"}
try:
params = {
"vnet": vnet,
"zone": zone,
}
if tag is not None:
params["tag"] = tag
if alias:
params["alias"] = alias
if vlanaware:
params["vlanaware"] = 1
self.proxmox.cluster.sdn.vnets.post(**params)
logger.info(f"Created SDN vnet: {vnet} in zone {zone}")
return {"success": True, "vnet": vnet}
except Exception as e:
logger.error(f"Error creating SDN vnet {vnet}: {e}")
return {"success": False, "error": str(e)}
def create_sdn_subnet(self, vnet: str, subnet: str, gateway: str = None,
snat: bool = False, dnszoneprefix: str = None) -> Dict:
"""
Create a subnet in a VNet.
Args:
vnet: VNet name
subnet: Subnet CIDR (e.g., "10.0.0.0/24")
gateway: Gateway IP for the subnet
snat: Enable SNAT for outgoing traffic
dnszoneprefix: DNS zone prefix
Returns:
Result dict with success status
"""
if not self.proxmox:
return {"success": False, "error": "Not connected"}
try:
params = {
"subnet": subnet,
"type": "subnet",
}
if gateway:
params["gateway"] = gateway
if snat:
params["snat"] = 1
if dnszoneprefix:
params["dnszoneprefix"] = dnszoneprefix
self.proxmox.cluster.sdn.vnets(vnet).subnets.post(**params)
logger.info(f"Created subnet {subnet} in vnet {vnet}")
return {"success": True, "subnet": subnet}
except Exception as e:
logger.error(f"Error creating subnet {subnet} in vnet {vnet}: {e}")
return {"success": False, "error": str(e)}