file

DefaultFileImportModule (FileImportModule)

Import an external file into a kiara session.

Source code in core/file.py

class DefaultFileImportModule(FileImportModule):
    """Import an external file into a kiara session."""

    _module_type_name = "import"

    def import_from__file_path__string(self, source: str) -> KiaraFile:

        file_model = KiaraFile.load_file(source)
        return file_model

LoadLocalFileModule (KiaraModule)

Load a file and its metadata.

This module does not read or load the content of a file, but contains the path to the local representation/version of the file so it can be read by a subsequent process.

Source code in core/file.py

class LoadLocalFileModule(KiaraModule):
    """Load a file and its metadata.

    This module does not read or load the content of a file, but contains the path to the local representation/version of the
    file so it can be read by a subsequent process.
    """

    # _config_cls = ImportLocalPathConfig
    _module_type_name = "load"

    def create_input_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:
        return {
            "base_path": {
                "type": "string",
                "doc": "The path to the base directory where the file is stored.",
            },
            "rel_path": {
                "type": "string",
                "doc": "The relative path of the file within the base directory.",
            },
        }

    def create_output_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:
        return {
            "file": {
                "type": "file",
                "doc": "A representation of the original file content in the kiara data registry.",
            }
        }

    def process(self, inputs: ValueSet, outputs: ValueSet) -> None:

        base_path = inputs.get_value_data("base_path")
        rel_path = inputs.get_value_data("rel_path")

        path = os.path.join(base_path, rel_path)

        file_model = KiaraFile.load_file(path)
        outputs.set_value("file", file_model)

create_input_schema(self)

Abstract method to implement by child classes, returns a description of the input schema of this module.

If returning a dictionary of dictionaries, the format of the return value is as follows (items with '*' are optional):

{ "[input_field_name]: { "type": "[value_type]", "doc*": "[a description of this input]", "optional*': [boolean whether this input is optional or required (defaults to 'False')] "[other_input_field_name]: { "type: ... ... }

Source code in core/file.py

def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:
    return {
        "base_path": {
            "type": "string",
            "doc": "The path to the base directory where the file is stored.",
        },
        "rel_path": {
            "type": "string",
            "doc": "The relative path of the file within the base directory.",
        },
    }

create_output_schema(self)

Abstract method to implement by child classes, returns a description of the output schema of this module.

If returning a dictionary of dictionaries, the format of the return value is as follows (items with '*' are optional):

{ "[output_field_name]: { "type": "[value_type]", "doc*": "[a description of this output]" "[other_input_field_name]: { "type: ... ... }

Source code in core/file.py

def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:
    return {
        "file": {
            "type": "file",
            "doc": "A representation of the original file content in the kiara data registry.",
        }
    }

StoreFileTypeModule (StoreValueTypeModule)

Save a file to disk.

Source code in core/file.py

class StoreFileTypeModule(StoreValueTypeModule):
    """Save a file to disk."""

    _module_type_name = "store"

    @classmethod
    def retrieve_supported_types(cls) -> typing.Union[str, typing.Iterable[str]]:
        return "file"

    def store_value(
        self, value: Value, base_path: str
    ) -> typing.Tuple[typing.Dict[str, typing.Any], typing.Any]:

        file_obj = value.get_value_data()

        file_name = file_obj.file_name
        full_target = os.path.join(base_path, file_name)

        os.makedirs(os.path.dirname(full_target), exist_ok=True)

        if os.path.exists(full_target):
            raise KiaraProcessingException(
                f"Can't save file, path already exists: {full_target}"
            )

        fm = file_obj.copy_file(full_target, is_onboarded=True)

        load_config = {
            "module_type": "file.load",
            "base_path_input_name": "base_path",
            "inputs": {"base_path": base_path, "rel_path": file_name},
            "output_name": "file",
        }
        return (load_config, fm)

store_value(self, value, base_path)

Save the value, and return the load config needed to load it again.

Source code in core/file.py

def store_value(
    self, value: Value, base_path: str
) -> typing.Tuple[typing.Dict[str, typing.Any], typing.Any]:

    file_obj = value.get_value_data()

    file_name = file_obj.file_name
    full_target = os.path.join(base_path, file_name)

    os.makedirs(os.path.dirname(full_target), exist_ok=True)

    if os.path.exists(full_target):
        raise KiaraProcessingException(
            f"Can't save file, path already exists: {full_target}"
        )

    fm = file_obj.copy_file(full_target, is_onboarded=True)

    load_config = {
        "module_type": "file.load",
        "base_path_input_name": "base_path",
        "inputs": {"base_path": base_path, "rel_path": file_name},
        "output_name": "file",
    }
    return (load_config, fm)