kiara.operations.serialize

SerializeValueModule

Base class for 'serialize' operations.

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 kiara/operations/serialize.py

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

    value_type: str = self.get_config_value("value_type")
    # serialization_type: str = self.get_config_value("serialization_type")

    return {
        "value_item": {
            "type": value_type,
            "doc": f"The '{value_type}' value to be serialized.",
        }
    }

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 kiara/operations/serialize.py

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

    value_type: str = self.get_config_value("value_type")

    return {
        "value_info": {
            "type": "value_info",
            "doc": "Information about the (original) serialized value (can be used to re-constitute the value, incl. its original id).",
        },
        "deserialize_config": {
            "type": "deserialize_config",
            "doc": f"The config to use to deserialize the value of type '{value_type}'.",
        },
    }

retrieve_module_profiles(kiara) classmethod

Retrieve a collection of profiles (pre-set module configs) for this kiara module type.

This is used to automatically create generally useful operations (incl. their ids).

Source code in kiara/operations/serialize.py

@classmethod
def retrieve_module_profiles(
    cls, kiara: "Kiara"
) -> typing.Mapping[str, typing.Union[typing.Mapping[str, typing.Any], Operation]]:

    all_profiles: typing.Dict[
        str, typing.Dict[str, typing.Dict[str, typing.Any]]
    ] = {}

    value_type: str = cls.get_value_type()

    if value_type not in kiara.type_mgmt.value_type_names:
        log_message(
            f"Ignoring serialization operation for source type '{value_type}': type not available"
        )
        return {}

    for serialization_type in cls.get_supported_serialization_types():

        mod_conf = {
            "value_type": value_type,
            "serialization_type": serialization_type,
        }

        op_config = {
            "module_type": cls._module_type_id,  # type: ignore
            "module_config": mod_conf,
            "doc": f"Serialize value of type '{value_type}' to '{serialization_type}'.",
        }
        key = f"serialize.{value_type}.as.{serialization_type}"
        if key in all_profiles.keys():
            raise Exception(f"Duplicate profile key: {key}")
        all_profiles[key] = op_config

    return all_profiles

SerializeValueModuleConfig pydantic-model

serialization_type: str pydantic-field required

The type of the converted value.

value_type: str pydantic-field required

The type of the source value.

SerializeValueOperationType

Operations that serialize data into formats that can be used for data exchange.

NOT USED YET

get_operations_for_value_type(self, value_type)

Find all operations that serialize the specified type.

The result dict uses the serialization type as key, and the operation itself as value.

Source code in kiara/operations/serialize.py

def get_operations_for_value_type(
    self, value_type: str
) -> typing.Dict[str, Operation]:
    """Find all operations that serialize the specified type.

    The result dict uses the serialization type as key, and the operation itself as value.
    """

    result: typing.Dict[str, Operation] = {}
    for o_id, op in self.operations.items():
        source_type = op.module_config["value_type"]
        if source_type == value_type:
            target_type = op.module_config["serialization_type"]
            if target_type in result.keys():
                raise Exception(
                    f"Multiple operations to serialize '{source_type}' to {target_type}"
                )
            result[target_type] = op

    return result