Skip to content

kiara_modules.core.bytes

LoadBytesModule

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/bytes/__init__.py 
def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {
        "base_path": {
            "type": "string",
            "doc": "The base path to the file to read.",
        },
        "rel_path": {
            "type": "string",
            "doc": "The relative path of the file, within the base path.",
        },
    }

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/bytes/__init__.py 
def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {"bytes": {"type": "bytes", "doc": "The content of the file."}}

StoreBytesTypeModule

store_value(self, value, base_path)

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

Source code in core/bytes/__init__.py 
def store_value(self, value: Value, base_path: str) -> typing.Dict[str, typing.Any]:

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

    if os.path.exists(path):
        raise KiaraProcessingException(
            f"Can't write bytes, target path already exists: {path}"
        )

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

    bytes = value.get_value_data()

    with open(path, "wb") as f:
        f.write(bytes)

    load_config = {
        "module_type": "bytes.load",
        "inputs": {"base_path": base_path, "rel_path": BYTES_SAVE_FILE_NAME},
        "output_name": "bytes",
    }
    return load_config