from typing import TYPE_CHECKING
if TYPE_CHECKING:
from render import Render
from storage import IxStorage
try:
from .error import RenderError
from .formatter import merge_dicts_no_overwrite
from .volume_mount_types import BindMountType, VolumeMountType
from .volume_sources import HostPathSource, IxVolumeSource, CifsSource, NfsSource, VolumeSource
except ImportError:
from error import RenderError
from formatter import merge_dicts_no_overwrite
from volume_mount_types import BindMountType, VolumeMountType
from volume_sources import HostPathSource, IxVolumeSource, CifsSource, NfsSource, VolumeSource
class VolumeMount:
def __init__(self, render_instance: "Render", mount_path: str, config: "IxStorage"):
self._render_instance = render_instance
self.mount_path: str = mount_path
storage_type: str = config.get("type", "")
if not storage_type:
raise RenderError("Expected [type] to be set for volume mounts.")
match storage_type:
case "host_path":
spec_type = "bind"
mount_config = config.get("host_path_config")
if mount_config is None:
raise RenderError("Expected [host_path_config] to be set for [host_path] type.")
mount_type_specific_definition = BindMountType(self._render_instance, mount_config).render()
source = HostPathSource(self._render_instance, mount_config).get()
case "ix_volume":
spec_type = "bind"
mount_config = config.get("ix_volume_config")
if mount_config is None:
raise RenderError("Expected [ix_volume_config] to be set for [ix_volume] type.")
mount_type_specific_definition = BindMountType(self._render_instance, mount_config).render()
source = IxVolumeSource(self._render_instance, mount_config).get()
case "nfs":
spec_type = "volume"
mount_config = config.get("nfs_config")
if mount_config is None:
raise RenderError("Expected [nfs_config] to be set for [nfs] type.")
mount_type_specific_definition = VolumeMountType(self._render_instance, mount_config).render()
source = NfsSource(self._render_instance, mount_config).get()
case "cifs":
spec_type = "volume"
mount_config = config.get("cifs_config")
if mount_config is None:
raise RenderError("Expected [cifs_config] to be set for [cifs] type.")
mount_type_specific_definition = VolumeMountType(self._render_instance, mount_config).render()
source = CifsSource(self._render_instance, mount_config).get()
case "volume":
spec_type = "volume"
mount_config = config.get("volume_config")
if mount_config is None:
raise RenderError("Expected [volume_config] to be set for [volume] type.")
mount_type_specific_definition = VolumeMountType(self._render_instance, mount_config).render()
source = VolumeSource(self._render_instance, mount_config).get()
case "temporary":
spec_type = "volume"
mount_config = config.get("volume_config")
if mount_config is None:
raise RenderError("Expected [volume_config] to be set for [temporary] type.")
mount_type_specific_definition = VolumeMountType(self._render_instance, mount_config).render()
source = VolumeSource(self._render_instance, mount_config).get()
case "anonymous":
spec_type = "volume"
mount_config = config.get("volume_config") or {}
mount_type_specific_definition = VolumeMountType(self._render_instance, mount_config).render()
source = None
case _:
raise RenderError(f"Storage type [{storage_type}] is not supported for volume mounts.")
common_spec = {"type": spec_type, "target": self.mount_path, "read_only": config.get("read_only", False)}
if source is not None:
common_spec["source"] = source
self._render_instance.volumes.add_volume(source, storage_type, mount_config) # type: ignore
self.volume_mount_spec = merge_dicts_no_overwrite(common_spec, mount_type_specific_definition)
def render(self) -> dict:
return self.volume_mount_spec