apps/library/2.1.49/volume_sources.py

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from render import Render
    from storage import IxStorageHostPathConfig, IxStorageIxVolumeConfig, IxStorageVolumeConfig

try:
    from .error import RenderError
    from .formatter import get_hashed_name_for_volume
    from .validations import valid_fs_path_or_raise, allowed_fs_host_path_or_raise
except ImportError:
    from error import RenderError
    from formatter import get_hashed_name_for_volume
    from validations import valid_fs_path_or_raise, allowed_fs_host_path_or_raise


class HostPathSource:
    def __init__(self, render_instance: "Render", config: "IxStorageHostPathConfig"):
        self._render_instance = render_instance
        self.source: str = ""

        if not config:
            raise RenderError("Expected [host_path_config] to be set for [host_path] type.")

        path = ""
        if config.get("acl_enable", False):
            acl_path = config.get("acl", {}).get("path")
            if not acl_path:
                raise RenderError("Expected [host_path_config.acl.path] to be set for [host_path] type.")
            path = valid_fs_path_or_raise(acl_path)
        else:
            path = valid_fs_path_or_raise(config.get("path", ""))

        path = path.rstrip("/")
        self.source = allowed_fs_host_path_or_raise(path)

    def get(self):
        return self.source


class IxVolumeSource:
    def __init__(self, render_instance: "Render", config: "IxStorageIxVolumeConfig"):
        self._render_instance = render_instance
        self.source: str = ""

        if not config:
            raise RenderError("Expected [ix_volume_config] to be set for [ix_volume] type.")
        dataset_name = config.get("dataset_name")
        if not dataset_name:
            raise RenderError("Expected [ix_volume_config.dataset_name] to be set for [ix_volume] type.")

        ix_volumes = self._render_instance.values.get("ix_volumes", {})
        if dataset_name not in ix_volumes:
            available = ", ".join(ix_volumes.keys())
            raise RenderError(
                f"Expected the key [{dataset_name}] to be set in [ix_volumes] for [ix_volume] type. "
                f"Available keys: [{available}]."
            )

        path = valid_fs_path_or_raise(ix_volumes[dataset_name].rstrip("/"))
        self.source = allowed_fs_host_path_or_raise(path, True)

    def get(self):
        return self.source


class CifsSource:
    def __init__(self, render_instance: "Render", config: dict):
        self._render_instance = render_instance
        self.source: str = ""

        if not config:
            raise RenderError("Expected [cifs_config] to be set for [cifs] type.")
        self.source = get_hashed_name_for_volume("cifs", config)

    def get(self):
        return self.source


class NfsSource:
    def __init__(self, render_instance: "Render", config: dict):
        self._render_instance = render_instance
        self.source: str = ""

        if not config:
            raise RenderError("Expected [nfs_config] to be set for [nfs] type.")
        self.source = get_hashed_name_for_volume("nfs", config)

    def get(self):
        return self.source


class VolumeSource:
    def __init__(self, render_instance: "Render", config: "IxStorageVolumeConfig"):
        self._render_instance = render_instance
        self.source: str = ""

        if not config:
            raise RenderError("Expected [volume_config] to be set for [volume] type.")

        volume_name: str = config.get("volume_name", "")
        if not volume_name:
            raise RenderError("Expected [volume_config.volume_name] to be set for [volume] type.")

        self.source = volume_name

    def get(self):
        return self.source