generic

JsonSerializationConfig (StoreValueModuleConfig) pydantic-model

Source code in core/generic.py

class JsonSerializationConfig(StoreValueModuleConfig):

    options: int = Field(
        description="The options to use for the json serialization. Check https://github.com/ijl/orjson#quickstart for details.",
        default=orjson.OPT_NAIVE_UTC | orjson.OPT_SERIALIZE_NUMPY,
    )
    file_name: str = Field(
        description="The name of the serialized file.", default="dict.json"
    )

file_name: str pydantic-field

The name of the serialized file.

options: int pydantic-field

The options to use for the json serialization. Check https://github.com/ijl/orjson#quickstart for details.

RestoreFromJsonDictModule (KiaraModule)

Source code in core/generic.py

class RestoreFromJsonDictModule(KiaraModule):

    _module_type_name = "restore_from_json"

    def create_input_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:

        return {
            "base_path": {
                "type": "string",
                "doc": "The folder that contains the serialized dict.",
            },
            "file_name": {
                "type": "string",
                "doc": "The file name of the serialized dict.",
            },
        }

    def create_output_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:

        return {"value_item": {"type": "dict", "doc": "The deserialized dict value."}}

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

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

        full_path = os.path.join(base_path, file_name)

        if not os.path.exists(full_path):
            raise KiaraProcessingException(
                f"Can't deserialize dict, path to file does not exist: {full_path}"
            )

        if not os.path.isfile(os.path.realpath(full_path)):
            raise KiaraProcessingException(
                f"Can't deserialize dict, path is not a file: {full_path}"
            )

        with open(full_path, "r") as f:
            content = f.read()

        data = orjson.loads(content)
        outputs.set_value("value_item", data)

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

def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {
        "base_path": {
            "type": "string",
            "doc": "The folder that contains the serialized dict.",
        },
        "file_name": {
            "type": "string",
            "doc": "The file name of the serialized dict.",
        },
    }

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

def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {"value_item": {"type": "dict", "doc": "The deserialized dict value."}}

RestoreScalarModule (KiaraModule)

Utility module, only used internally.

Source code in core/generic.py

class RestoreScalarModule(KiaraModule):
    """Utility module, only used internally."""

    _module_type_name = "restore_scalar"
    _config_cls = RestoreScalarModuleConfig

    def create_input_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:

        return {
            "scalar_data": {
                "type": self.get_config_value("value_type"),
                "doc": "The scalar value.",
            }
        }

    def create_output_schema(
        self,
    ) -> typing.Mapping[
        str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
    ]:

        return {
            "value_item": {
                "type": self.get_config_value("value_type"),
                "doc": "The loaded item.",
            }
        }

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

        data = inputs.get_value_obj("scalar_data")
        outputs.set_value("value_item", data)

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

def create_input_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {
        "scalar_data": {
            "type": self.get_config_value("value_type"),
            "doc": "The scalar value.",
        }
    }

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

def create_output_schema(
    self,
) -> typing.Mapping[
    str, typing.Union[ValueSchema, typing.Mapping[str, typing.Any]]
]:

    return {
        "value_item": {
            "type": self.get_config_value("value_type"),
            "doc": "The loaded item.",
        }
    }

RestoreScalarModuleConfig (ModuleTypeConfigSchema) pydantic-model

Source code in core/generic.py

class RestoreScalarModuleConfig(ModuleTypeConfigSchema):

    value_type: str = Field(description="The value type of the scalar to load.")

value_type: str pydantic-field required

The value type of the scalar to load.

StoreScalarModule (StoreValueTypeModule)

Source code in core/generic.py

class StoreScalarModule(StoreValueTypeModule):

    _module_type_name = "store"

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

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

        data = value.get_value_data()

        load_config = {
            "module_type": "generic.restore_scalar",
            "module_config": {"value_type": self.get_config_value("value_type")},
            "base_path_input_name": None,
            "inputs": {"scalar_data": data},
            "output_name": "value_item",
        }

        return load_config

store_value(self, value, base_path)

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

Source code in core/generic.py

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

    data = value.get_value_data()

    load_config = {
        "module_type": "generic.restore_scalar",
        "module_config": {"value_type": self.get_config_value("value_type")},
        "base_path_input_name": None,
        "inputs": {"scalar_data": data},
        "output_name": "value_item",
    }

    return load_config

StoreScalarModuleConfig (ModuleTypeConfigSchema) pydantic-model

Source code in core/generic.py

class StoreScalarModuleConfig(ModuleTypeConfigSchema):

    value_type: str = Field(description="The value type of the scalar to store.")

value_type: str pydantic-field required

The value type of the scalar to store.