deevirt_schema/api/vm.py

from typing import List, Tuple, Optional
from enum import Enum, IntEnum
from uuid import UUID, uuid4
from pydantic_xml import element, attr, wrapped
from pydantic import UUID4, Field, field_serializer, ConfigDict, ValidationInfo, field_validator
from pydantic import BaseModel as _BaseModel
from pydantic_core import core_schema as cs
import libvirt

from pydantic_core import core_schema
from pydantic.json_schema import GetJsonSchemaHandler, JsonSchemaValue

from deevirt_common.lib.virt import vm as domain
from deevirt_common.schemas import datacenter_schema, storage_schema, network_schema, libvirt

class BaseModel(_BaseModel):
    class Config:
        @staticmethod
        def json_schema_extra(schema: dict, _):
            props = {}
            for k, v in schema.get('properties', {}).items():
                if not v.get("hidden", False):
                    props[k] = v
            schema["properties"] = props


""" Common """
class VMResource(BaseModel):
    """Experimental"""
    name:Optional[str] = Field(default=None, description="Nom du pool")

class VMVCPU(BaseModel):
    value:int = Field(description="Nombre de VCPU")
    dedicated:Optional[bool] = Field(default=False, description="VCPU dédié")
    hotplug:Optional[bool] = Field(default=False, description="Ajout de VCPU à chaud.")

class VMMemory(BaseModel):
    unit:Optional[libvirt.UnitEnum] = Field(default=libvirt.UnitEnum.GiB, description="Unité de mesure.")
    value:int = Field(description="Taille de la mémoire.")
    hotplug:Optional[bool] = Field(default=False,description="Ajout à chaud de la mémoire.")

class VMOSBootMenu(BaseModel):
    enable:Optional[bool] = Field(default=False,description="Activation du menu de démarrage.")
    delay:Optional[int] = Field(default=0, description="Delai maximal pour entrer dans le menu démarrage en millisecondes")

class VMOSBootOrder(BaseModel):
    dev:Optional[libvirt.OSBoot.bootEnum] = Field(default=None, description="Périphérique.")

class VMOSInfo(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    name:str = Field(alias="os.name", serialization_alias="name")
    version:str  = Field(alias="os.version-id", serialization_alias="version")
    machine:str  = Field(alias="os.machine", serialization_alias="architecture")
    kernel:str  = Field(alias="os.kernel-release", serialization_alias="kernel")

class VMOS(BaseModel):
    firmware:libvirt.OS.firmwareEnum = Field(default=libvirt.OS.firmwareEnum.bios, description="Microprogramme")
    enrolled_keys:Optional[bool] = Field(default=None, description="Vérification de la signature du chargeur de démarrage")
    secure_boot:Optional[bool] = Field(default=None, description="Démarrage sécurisé")
    boot_order:Optional[List[libvirt.OSBoot.bootEnum]] = Field(default=None, description="Ordre de démarrage des périphériques")
    boot_menu:VMOSBootMenu = Field(default=VMOSBootMenu(), description="Menu de démarrage")

class VMClock(BaseModel):
    class offsetEnum(str, Enum):
        utc = "utc"
        localtime = "localtime"
        timezone = "timezone"
        variable = "variable"
        absolute = "absolute"
    
    offset:offsetEnum = Field(default=offsetEnum.utc, description="Sélection de l'offset ?")

class VMEvent(BaseModel):
    class on_poweroffEnum(str, Enum):
        destroy = "destroy"
        restart = "restart"
        preserve = "preverse"

    class on_rebootEnum(str, Enum):
        restart = "restart"
        preserve = "preverse"

    class on_crashEnum(str, Enum):
        destroy = "destroy"
        restart = "restart"
        preserve = "preverse"

    class on_failurelockEnum(str, Enum):
        poweroff = "poweroff"
        restart = "restart"
        pause = "pause"
        ignore = "ignore"

    on_poweroff:on_poweroffEnum = Field(default=on_poweroffEnum.destroy, description="Action lors d'un poweroff")
    on_reboot:on_rebootEnum = Field(default=on_rebootEnum.restart, description="Action lors d'un redémarrage")
    on_crash:on_crashEnum = Field(default=on_crashEnum.restart, description="Action lors d'un crash de la VM")
    on_failurelock:on_failurelockEnum = element(default=on_failurelockEnum.ignore, description="A tester")

class VMDevicesDisk(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    pool_id:UUID4 = Field()
    id:UUID4 = Field()
    size:int = Field()
    unit:libvirt.UnitEnum = Field(default=libvirt.UnitEnum.GiB)
    bus:libvirt.DevicesDiskTarget.busEnum = Field(default=libvirt.DevicesDiskTarget.busEnum.scsi)

class VMDevicesCDROMSource(BaseModel):
    class protocolEnum(str, Enum):
        http = "http"
        https = "https"
        ftp = "ftp"
        ftps = "ftps"

    protocol:protocolEnum = attr(default=protocolEnum.https)
    name:str = attr()

class VMDevicesCDROM(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    device:libvirt.DevicesDisk.deviceEnum = Field(default=libvirt.DevicesDisk.deviceEnum.cdrom, json_schema_extra={"hidden": True})
    source:VMDevicesCDROMSource = element(default=None)

class VMDevicesInterfaceGuest(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    addr:str = Field()

class VMDevicesInterface(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    switch_id:UUID4
    id:UUID4
    mac_addr:str = Field()

class VMGraphics(BaseModel):
    port:int = Field(default=-1)
    autoport:bool = Field(default=False)
    websocket:int = Field(default=-1)
    password:Optional[str] = Field(default=None)
    keymap:libvirt.Graphics.keymapEnum = Field(default=libvirt.Graphics.keymapEnum.fr)
    power_control:bool = Field(default=False)

class VMAudio(BaseModel):
    pass

class VMVideoModel(BaseModel):
    type:libvirt.VideoModel.typeEnum = Field(default=libvirt.VideoModel.typeEnum.cirrus)
    vram:int = Field(default=16384)
    primary:libvirt.YesNoEnum = Field(default=libvirt.YesNoEnum.YES)

class VMVideo(BaseModel):
    model:VMVideoModel = element() 

class VMWatchdog(BaseModel):
    model:libvirt.Watchdog.modelEnum = Field(default=libvirt.Watchdog.modelEnum.itco)
    action:libvirt.Watchdog.actionEnum = Field(default=libvirt.Watchdog.actionEnum.reset)

class VMTPM(BaseModel):
    model:libvirt.TPM.modelEnum = attr(default=libvirt.TPM.modelEnum.tpm_tis)
    version:libvirt.TPMBackend.versionEnum  = Field(default=libvirt.TPMBackend.versionEnum.v20)

class VMDevices(BaseModel):
    disks:Optional[List[VMDevicesDisk]] = Field(default=None, description="Disques virtuel")
    cdrom:VMDevicesCDROM = Field(default=None, description="CDROM virtuel")
    interfaces:Optional[List[VMDevicesInterface]] = Field(default=None, description="Interfaces réseau")
    graphics:Optional[VMGraphics] = Field(default=None, description="Console VNC")
    audio:VMAudio = Field(default=None, description="Carte Audio")
    video:VMVideo = Field(default=None, description="Carte Vidéo")
    watchdog:VMWatchdog = Field(default=None, description="Chien de garde")
    tpm:VMTPM = Field(default=None, description="Puce de sécurité")

class VM(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    # General metadata
    id:UUID4 = Field()
    datacenter_id:Optional[UUID4] = Field(default=None)
    name:str = Field()
    description:Optional[str] = Field(default=None)

    # CPU Allocation
    vcpu:VMVCPU = Field()

    # Memory Allocation
    memory:VMMemory = Field()

    # Pool of resource (Experimental)
    resource:VMResource = Field(default=VMResource(), description="Pool de resources")

    # BIOS BootLoader
    os:VMOS = Field(default=VMOS(), description="Options VM: Options de démarrage.")

    # Clock configuration
    clock:VMClock = Field(default=VMClock(), description="Options VM: Gestion de l'horloge")

    # Event configuration
    event:VMEvent = Field(default=VMEvent(), description="Options VM: Gestion des évènements")

    # Devices
    devices:VMDevices = Field(default=VMDevices(), description="Matériel virtuel: Périphériques")

"""VM LIST GET"""
class VMs(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    class Datacenter(BaseModel):
        model_config = ConfigDict(from_attributes=True)
        
        id:UUID4
        name:str

    id:UUID4
    name:str
    state:int
    datacenter:Optional[Datacenter] = Field(default=[])

"""VM GET"""
class VMGetVCPU(VMVCPU):
    max:Optional[int] = Field(default=None, description="Maximum de VCPUs")

class VMGetOS(VMOS):
    guest_info:Optional[VMOSInfo] = Field(default=None, description="Information de l'OS client")

class VMGetDevicesDisk(VMDevicesDisk):
    allocated:float = Field(default=None)

class VMGetDevicesInterface(VMDevicesInterface):
    guest_info:Optional[List[VMDevicesInterfaceGuest]] = Field(default=None, description="Information de configuration de la carte réseau.")

class VMGetDevices(VMDevices):
    disks:Optional[List[VMGetDevicesDisk]] = Field(default=None, description="Disques virtuel")
    interfaces:Optional[List[VMGetDevicesInterface]] = Field(default=None, description="Interfaces réseau")

class VMGet(VM):
    vcpu:VMGetVCPU = Field()
    gen_id:Optional[UUID4] = Field(default=None)
    os:VMGetOS = Field(default=VMGetOS(), description="Options VM: Options de démarrage.")
    devices:VMGetDevices = Field(default=VMGetDevices(), description="Matériel virtuel: Périphériques")

"""VM POST"""
class VMCreate(VM):
    id:UUID4 = Field(default_factory=uuid4)
    datacenter_id:UUID4 = Field()

"""VM DELETE"""
class VMDelete(BaseModel):
    id:UUID4 = Field()

class VMHardDisk(BaseModel):
    pass

class VMNetInterfaceCreate(BaseModel):
    pass