from sio3pack.exceptions import ParsingFailedOn, WorkflowParsingError
from sio3pack.workflow.object import Object
[docs]
class Filesystem:
"""
A base class to represent a filesystem.
:param int id: The id of the filesystem in the task.
"""
_required_keys = []
def __init__(self, id: int = None):
"""
Represent a filesystem.
:param id: The id of the filesystem.
"""
self.id = id
def _set_id(self, id: int):
"""
Set the id of the filesystem. Should only be used by the FilesystemManager.
:param int id: The id of the filesystem.
"""
self.id = id
[docs]
@classmethod
def from_json(cls, data: dict, id: int, workflow: "Workflow"):
"""
Create a new filesystem from a dictionary.
:param dict data: The dictionary to create the filesystem from.
:param int id: The id of the filesystem.
:param Workflow workflow: The workflow the filesystem belongs to.
"""
for key in cls._required_keys:
if key not in data:
raise WorkflowParsingError(
"Parsing filesystem failed.",
ParsingFailedOn.FILESYSTEM,
extra_msg=f"Missing required key '{key}' in filesystem definition.",
data={"filesystem_index": id},
)
[docs]
def to_json(self) -> dict:
"""
Convert the filesystem to a dictionary.
:return: The dictionary representation of the filesystem.
"""
return NotImplementedError()
[docs]
def replace_templates(self, replacements: dict[str, str]):
"""
Replace strings in the task with the given replacements.
:param replacements: The replacements to make.
"""
pass
[docs]
class ImageFilesystem(Filesystem):
"""
A class to represent an image filesystem.
An image can be for example a g++ compilator, a python interpreter, etc.
:param int id: The id of the image filesystem.
:param str image: The image to use.
:param str path: The path to the image. If None, the path is "".
"""
_required_keys = ["image", "path"]
def __init__(self, image: str, path: str = None, id: int = None):
"""
Represent an image filesystem.
:param image: The image to use.
:param path: The path to the image. If None, the path is ""
"""
super().__init__(id)
self.image = image
self.path = path or ""
def __str__(self):
return f'<ImageFilesystem {self.image} path="{self.path}">'
[docs]
@classmethod
def from_json(cls, data: dict, id: int, workflow: "Workflow") -> "ImageFilesystem":
"""
Create a new image filesystem from a dictionary.
:param dict data: The dictionary to create the image filesystem from.
:param id id: The id of the image filesystem.
:param Workflow workflow: The workflow the image filesystem belongs to.
"""
super().from_json(data, id, workflow)
return cls(data["image"], data["path"], id)
[docs]
def to_json(self) -> dict:
"""
Convert the image filesystem to a dictionary.
:return: The dictionary representation of the image filesystem.
"""
return {"type": "image", "image": self.image, "path": self.path}
[docs]
class EmptyFilesystem(Filesystem):
_required_keys = []
def __init__(self, id: int = None):
"""
Represent an empty filesystem. Can be used as tmpfs.
:param id: The id of the empty filesystem.
"""
super().__init__(id)
def __str__(self):
return "<EmptyFilesystem>"
[docs]
@classmethod
def from_json(cls, data: dict, id: int, workflow: "Workflow"):
"""
Create a new empty filesystem from a dictionary.
:param data: The dictionary to create the empty filesystem from.
:param id: The id of the empty filesystem.
:param workflow: The workflow the empty filesystem belongs to.
"""
super().from_json(data, id, workflow)
return cls(id)
[docs]
def to_json(self) -> dict:
"""
Convert the empty filesystem to a dictionary.
"""
return {"type": "empty"}
[docs]
class ObjectFilesystem(Filesystem):
_required_keys = ["handle"]
def __init__(self, object: Object, id: int = None):
"""
Represent an object filesystem.
:param object: The object to use.
:param workflow: The workflow the object belongs to.
"""
super().__init__(id)
self.object = object
def __str__(self):
return f"<ObjectFilesystem {self.object.handle}>"
[docs]
@classmethod
def from_json(cls, data: dict, id: int, workflow: "Workflow"):
"""
Create a new object filesystem from a dictionary.
:param data: The dictionary to create the object filesystem from.
:param id: The id of the object filesystem.
:param workflow: The workflow the object filesystem belongs to.
"""
super().from_json(data, id, workflow)
return cls(workflow.objects_manager.get_or_create_object(data["handle"]), id)
[docs]
def to_json(self) -> dict:
"""
Convert the object filesystem to a dictionary.
"""
return {
"type": "object",
"handle": self.object.handle,
}
[docs]
def replace_templates(self, replacements: dict[str, str]):
"""
Replace strings in the task with the given replacements.
:param replacements: The replacements to make.
"""
self.object.replace_templates(replacements)
[docs]
class FilesystemManager:
"""
A class to manage filesystems.
:param Task task: The task the filesystem manager belongs to.
:param list[Filesystem] filesystems: The list of filesystems.
"""
def __init__(self, task: "Task"):
"""
Create a new filesystem manager.
:param Task task: The task the filesystem manager belongs to.
"""
self.filesystems: list[Filesystem] = []
self.id = 0
self.task = task
[docs]
def from_json(self, data: list[dict], workflow: "Workflow"):
"""
Create filesystems from a list of dictionaries.
:param list[dict] data: The list of dictionaries to create the filesystems from.
:param Workflow workflow: The workflow the filesystems belong to.
"""
for i, fs in enumerate(data):
if "type" not in fs:
raise WorkflowParsingError(
"Parsing filesystem failed.",
ParsingFailedOn.FILESYSTEM,
extra_msg="Missing 'type' key in filesystem definition.",
data={"filesystem_index": i},
)
if fs["type"] == "image":
self.filesystems.append(ImageFilesystem.from_json(fs, self.id, workflow))
elif fs["type"] == "empty":
self.filesystems.append(EmptyFilesystem.from_json(fs, self.id, workflow))
elif fs["type"] == "object":
self.filesystems.append(ObjectFilesystem.from_json(fs, self.id, workflow))
else:
raise WorkflowParsingError(
"Parsing filesystem failed.",
ParsingFailedOn.FILESYSTEM,
extra_msg=f"Unknown filesystem type '{fs['type']}' in filesystem definition.",
data={"filesystem_index": i},
)
self.id += 1
[docs]
def to_json(self) -> list[dict]:
"""
Convert the filesystems to a list of dictionaries.
"""
return [fs.to_json() for fs in self.filesystems]
[docs]
def get_by_id(self, id: int) -> Filesystem:
"""
Get a filesystem by id.
:param id: The id of the filesystem.
"""
return self.filesystems[id]
[docs]
def add(self, filesystem: Filesystem):
"""
Add a filesystem to the manager.
:param filesystem: The filesystem to add.
"""
filesystem._set_id(self.id)
self.filesystems.append(filesystem)
self.id += 1
[docs]
def replace_templates(self, replacements: dict[str, str]):
"""
Replace strings in the task with the given replacements.
:param replacements: The replacements to make.
"""
for fs in self.filesystems:
fs.replace_templates(replacements)
[docs]
def len(self) -> int:
"""
Get the number of filesystems.
"""
return len(self.filesystems)
[docs]
def has_by_id(self, id: int) -> bool:
"""
Check if a filesystem with the given id exists.
:param id: The id of the filesystem to check.
:return: True if the filesystem exists, False otherwise.
"""
return 0 <= id < len(self.filesystems) and self.filesystems[id] is not None