volumes.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. from typing import TYPE_CHECKING
  2. if TYPE_CHECKING:
  3. from render import Render
  4. try:
  5. from .error import RenderError
  6. from .storage import IxStorageVolumeLikeConfigs
  7. from .volume_types import NfsVolume, CifsVolume, DockerVolume
  8. except ImportError:
  9. from error import RenderError
  10. from storage import IxStorageVolumeLikeConfigs
  11. from volume_types import NfsVolume, CifsVolume, DockerVolume
  12. class Volumes:
  13. def __init__(self, render_instance: "Render"):
  14. self._render_instance = render_instance
  15. self._volumes: dict[str, Volume] = {}
  16. def add_volume(self, source: str, storage_type: str, config: "IxStorageVolumeLikeConfigs"):
  17. # This method can be called many times from the volume mounts
  18. # Only add the volume if it is not already added, but dont raise an error
  19. if source == "":
  20. raise RenderError(f"Volume source [{source}] cannot be empty")
  21. if source in self._volumes:
  22. return
  23. self._volumes[source] = Volume(self._render_instance, storage_type, config)
  24. def has_volumes(self) -> bool:
  25. return bool(self._volumes)
  26. def render(self):
  27. return {name: v.render() for name, v in sorted(self._volumes.items()) if v.render() is not None}
  28. class Volume:
  29. def __init__(
  30. self,
  31. render_instance: "Render",
  32. storage_type: str,
  33. config: "IxStorageVolumeLikeConfigs",
  34. ):
  35. self._render_instance = render_instance
  36. self.volume_spec: dict | None = {}
  37. match storage_type:
  38. case "nfs":
  39. self.volume_spec = NfsVolume(self._render_instance, config).get() # type: ignore
  40. case "cifs":
  41. self.volume_spec = CifsVolume(self._render_instance, config).get() # type: ignore
  42. case "volume" | "temporary":
  43. self.volume_spec = DockerVolume(self._render_instance, config).get() # type: ignore
  44. case _:
  45. self.volume_spec = None
  46. def render(self):
  47. return self.volume_spec