kiara.operations.create_value

CreateValueModule

Base class for 'create' value type 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/create_value.py

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

    source_profile = self.get_config_value("source_profile")

    source_config = dict(
        self._kiara.type_mgmt.get_type_config_for_data_profile(source_profile)
    )
    source_config["doc"] = f"The '{source_profile}' value to be converted."

    schema: typing.Dict[str, typing.Dict[str, typing.Any]] = {
        source_profile: source_config
    }
    if self.get_config_value("allow_none_input"):
        schema[source_config["type"]]["optional"] = True

    return schema

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

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

    source_profile = self.get_config_value("source_profile")

    target_type = self.get_config_value("target_type")

    schema: typing.Dict[str, typing.Dict[str, typing.Any]] = {
        target_type: {
            "type": target_type,
            "doc": f"The converted '{source_profile}' value as '{target_type}'.",
        }
    }
    if self.get_config_value("allow_none_input"):
        schema[target_type]["optional"] = True

    return schema

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

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

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

    # value_types: typing.Iterable[str] = cls.get_supported_value_types()
    # if "*" in value_types:
    #     value_types = kiara.type_mgmt.value_type_names
    #
    # for value_type in value_types:

    target_type = cls.get_target_value_type()

    if target_type not in kiara.type_mgmt.value_type_names:
        raise Exception(
            f"Can't assemble type convert operations for target type '{target_type}': type not available"
        )

    for source_profile in cls.get_source_value_profiles():

        mod_conf = {
            "source_profile": source_profile,
            "target_type": target_type,
        }

        op_config = {
            "module_type": cls._module_type_id,  # type: ignore
            "module_config": mod_conf,
            "doc": f"Create a '{target_type}' value from a '{source_profile}'.",
        }
        # key = f"{value_type}.convert_from.{target_type}"   # type: ignore
        # if key in all_metadata_profiles.keys():
        #     raise Exception(f"Duplicate profile key: {key}")
        # all_metadata_profiles[key] = op_config
        key = f"create.{target_type}.from.{source_profile}"
        if key in all_metadata_profiles.keys():
            raise Exception(f"Duplicate profile key: {key}")
        all_metadata_profiles[key] = op_config

    return all_metadata_profiles

CreateValueModuleConfig pydantic-model

allow_none_input: bool pydantic-field

Whether to allow 'none' source values, if one is encountered 'none' is returned.

source_profile: str pydantic-field required

The profile of the source value.

target_type: str pydantic-field required

The type of the value to convert to.

CreateValueOperationType

Operations that create values of specific types from values of certain value profiles.

In most cases, source profiles will be 'file' or 'file_bundle' values of some sort, and the created values will exhibit more inherent structure and stricter specs than their sources.

The 'create' operation type differs from 'import' in that it expects values that are already onboarded and are all information in the value is stored in a kiara data registry (in most cases the kiara data store).

get_operations_for_target_type(self, value_type)

Find all operations that transform to the specified type.

The result dict uses the source type of the conversion as key, and the operation itself as value.

Source code in kiara/operations/create_value.py

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

    The result dict uses the source type of the conversion as key, and the operation itself as value.
    """

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