kiara_modules.core.list

IncludedInListCheckModule

Check whether an element is in a list.

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/list.py

def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:
    inputs = {
        "list": {"type": "list", "doc": "The list."},
        "item": {
            "type": "any",
            "doc": "The element to check for inclusion in the list.",
        },
    }
    return inputs

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/list.py

def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:
    outputs = {
        "is_included": {
            "type": "boolean",
            "doc": "Whether the element is in the list, or not.",
        }
    }
    return outputs

StoreDictModule

store_value(self, value, base_path)

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

Source code in core/list.py

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

    import orjson

    options = self.get_config_value("options")
    file_name = self.get_config_value("file_name")
    json_str = orjson.dumps(value.get_value_data(), option=options)

    bp = Path(base_path)
    bp.mkdir(parents=True, exist_ok=True)

    full_path = bp / file_name
    full_path.write_bytes(json_str)

    load_config = {
        "module_type": "generic.restore_from_json",
        "base_path_input_name": "base_path",
        "inputs": {
            "base_path": base_path,
            "file_name": self.get_config_value("file_name"),
        },
        "output_name": "value_item",
    }

    return load_config