from typing import TYPE_CHECKING
if TYPE_CHECKING:
from container import Container
from render import Render
from storage import IxStorage
try:
from .error import RenderError
from .validations import valid_fs_path_or_raise, valid_octal_mode_or_raise
except ImportError:
from error import RenderError
from validations import valid_fs_path_or_raise, valid_octal_mode_or_raise
class Tmpfs:
def __init__(self, render_instance: "Render", container_instance: "Container"):
self._render_instance = render_instance
self._container_instance = container_instance
self._tmpfs: dict = {}
def add(self, mount_path: str, config: "IxStorage"):
mount_path = valid_fs_path_or_raise(mount_path)
if self.is_defined(mount_path):
raise RenderError(f"Tmpfs mount path [{mount_path}] already added")
if self._container_instance.storage.is_defined(mount_path):
raise RenderError(f"Tmpfs mount path [{mount_path}] already used for another volume mount")
mount_config = config.get("tmpfs_config", {})
size = mount_config.get("size", None)
mode = mount_config.get("mode", None)
uid = mount_config.get("uid", None)
gid = mount_config.get("gid", None)
if size is not None:
if not isinstance(size, int):
raise RenderError(f"Expected [size] to be an integer for [tmpfs] type, got [{size}]")
if not size > 0:
raise RenderError(f"Expected [size] to be greater than 0 for [tmpfs] type, got [{size}]")
# Convert Mebibytes to Bytes
size = size * 1024 * 1024
if mode is not None:
mode = valid_octal_mode_or_raise(mode)
if uid is not None and not isinstance(uid, int):
raise RenderError(f"Expected [uid] to be an integer for [tmpfs] type, got [{uid}]")
if gid is not None and not isinstance(gid, int):
raise RenderError(f"Expected [gid] to be an integer for [tmpfs] type, got [{gid}]")
self._tmpfs[mount_path] = {}
if size is not None:
self._tmpfs[mount_path]["size"] = str(size)
if mode is not None:
self._tmpfs[mount_path]["mode"] = str(mode)
if uid is not None:
self._tmpfs[mount_path]["uid"] = str(uid)
if gid is not None:
self._tmpfs[mount_path]["gid"] = str(gid)
def is_defined(self, mount_path: str):
return mount_path in self._tmpfs
def has_tmpfs(self):
return bool(self._tmpfs)
def render(self):
result = []
for mount_path, config in self._tmpfs.items():
opts = sorted([f"{k}={v}" for k, v in config.items()])
result.append(f"{mount_path}:{','.join(opts)}" if opts else mount_path)
return sorted(result)