kiara.data.values.value_set

SlottedValueSet

__init__(self, items, read_only, check_for_sameness=False, title=None, kiara=None, registry=None) special

A ValueSet implementation that keeps a history of each fields value.

Parameters:

Name	Type	Description	Default
items	Mapping[str, Union[ValueSlot, Value]]	the value slots	required
read_only	bool	whether it is allowed to set new values to fields in this set	required
check_for_sameness	bool	whether a check should be performed that checks for equality of the new and old values, if equal, skip the update	False
title	Optional[str]	An optional title for this value set	None
kiara	Optional[Kiara]	the kiara context	None
registry	Optional[kiara.data.registry.DataRegistry]	the registry to use to register the values to	None

Source code in kiara/data/values/value_set.py

def __init__(
    self,
    items: typing.Mapping[str, typing.Union["ValueSlot", "Value"]],
    read_only: bool,
    check_for_sameness: bool = False,
    title: typing.Optional[str] = None,
    kiara: typing.Optional["Kiara"] = None,
    registry: typing.Optional[DataRegistry] = None,
):
    """A `ValueSet` implementation that keeps a history of each fields value.

    Arguments:
        items: the value slots
        read_only: whether it is allowed to set new values to fields in this set
        check_for_sameness: whether a check should be performed that checks for equality of the new and old values, if equal, skip the update
        title: An optional title for this value set
        kiara: the kiara context
        registry: the registry to use to register the values to

    """

    if not items:
        raise ValueError("Can't create ValueSet: no values provided")

    self._check_for_sameness: bool = check_for_sameness

    super().__init__(read_only=read_only, title=title, kiara=kiara)

    if registry is None:
        registry = self._kiara.data_registry
    self._registry: DataRegistry = registry

    _value_slots: typing.Dict[str, ValueSlot] = {}
    for item, value in items.items():

        if value is None:
            raise Exception(
                f"Can't create value set, item '{item}' does not have a value (yet)."
            )

        if item.startswith("_"):
            raise ValueError(f"Value name can't start with '_': {item}")
        if item in INVALID_VALUE_NAMES:
            raise ValueError(f"Invalid value name '{item}'.")

        if isinstance(value, Value):
            slot = self._registry.register_alias(value)
        elif isinstance(value, ValueSlot):
            slot = value
        else:
            raise TypeError(f"Invalid type: '{type(value)}'")

        _value_slots[item] = slot

    self._value_slots: typing.Dict[str, ValueSlot] = _value_slots

ValueSet

check_invalid(self)

Check whether the value set is invalid, if it is, return a description of what's wrong.

Source code in kiara/data/values/value_set.py

def check_invalid(self) -> typing.Optional[typing.Dict[str, str]]:
    """Check whether the value set is invalid, if it is, return a description of what's wrong."""

    invalid = {}
    for field_name, item in self.items():
        if not item.item_is_valid():
            if item.value_schema.is_required():
                if not item.is_set:
                    msg = "not set"
                elif item.is_none:
                    msg = "no value"
                else:
                    msg = "n/a"
            else:
                msg = "n/a"
            invalid[field_name] = msg
    return invalid

get_value_data_for_fields(self, *field_names, *, raise_exception_when_unset=False)

Return the data for a one or several fields of this ValueSet.

If a value is unset, by default 'None' is returned for it. Unless 'raise_exception_when_unset' is set to 'True', in which case an Exception will be raised (obviously).

Source code in kiara/data/values/value_set.py

def get_value_data_for_fields(
    self, *field_names: str, raise_exception_when_unset: bool = False
) -> typing.Dict[str, typing.Any]:
    """Return the data for a one or several fields of this ValueSet.

    If a value is unset, by default 'None' is returned for it. Unless 'raise_exception_when_unset' is set to 'True',
    in which case an Exception will be raised (obviously).
    """

    result: typing.Dict[str, typing.Any] = {}
    unset: typing.List[str] = []
    for k in field_names:
        v = self.get_value_obj(k)
        if not v.is_set:
            if raise_exception_when_unset:
                unset.append(k)
            else:
                result[k] = None
        else:
            data = v.get_value_data()
            result[k] = data

    if unset:
        raise Exception(
            f"Can't get data for fields, one or several of the requested fields are not set yet: {', '.join(unset)}."
        )

    return result

set_values(self, metadata=None, lineage=None, **values)

Batch set several values.

If metadata is provided, it is added to all values.

Source code in kiara/data/values/value_set.py

def set_values(
    self,
    metadata: typing.Optional[typing.Mapping[str, MetadataModel]] = None,
    lineage: typing.Optional[ValueLineage] = None,
    **values: typing.Any,
) -> typing.Mapping[str, typing.Union[Value, Exception]]:
    """Batch set several values.

    If metadata is provided, it is added to all values.
    """

    if self.is_read_only():
        raise Exception("Can't set values: this value set is read-only.")

    if not values:
        return {}

    invalid: typing.List[str] = []

    for k in values.keys():
        if k not in self.get_all_field_names():
            invalid.append(k)

    if invalid:
        raise ValueError(
            f"No field(s) with name(s) {', '.join(invalid)} available, valid names: {', '.join(self.get_all_field_names())}"
        )

    resolved_values = {}
    for field_name, data in values.items():
        if isinstance(data, str) and data.startswith("value:"):
            v = self._kiara.get_value(data)
            resolved_values[field_name] = v
        else:
            resolved_values[field_name] = data

    try:
        value_set_result = self._set_values(
            metadata=metadata, lineage=lineage, **resolved_values
        )
    except Exception as e:
        log_message(str(e))
        if is_debug():
            import traceback

            traceback.print_exc()
        raise e

    result: typing.Dict[str, typing.Union[Value, Exception]] = {}
    for field in values.keys():
        if isinstance(value_set_result[field], Exception):
            result[field] = value_set_result[field]  # type: ignore
        else:
            result[field] = self.get_value_obj(field)
    return result