kiara_modules.core.file

DefaultFileImportModule

Import an external file into a kiara session.

LoadLocalFileModule

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.

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

Save a file to disk.

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)