kiara.data.registry

BaseDataRegistry

Base class to extend if you want to write a kiara data registry.

This outlines the main methods that must be available for an entity that holds some values and their data, as well as their associated aliases. In most cases users will interact with subclasses of the more fully featured [DataRegistry][kiara.data.registry.DataRegistry] class, but there are cases where it will make sense to sub-class this base class directly (like for example read-only 'archive' data registries).

find_all_values_of_type(self, value_type)

Find all values of a certain type.

Source code in kiara/data/registry/__init__.py

@deprecated(
    reason="This will be removed soon in favor of a more generic way to query values."
)
def find_all_values_of_type(self, value_type: str) -> typing.Dict[str, Value]:
    """Find all values of a certain type."""

    result = {}
    for value_id in self.value_ids:
        value_obj = self.get_value_obj(value_item=value_id)
        if value_obj is None:
            raise Exception(f"No value registered for: {value_id}")
        if value_obj.type_name == value_type:  # type: ignore
            result[value_id] = value_obj
    return result

get_versions_for_alias(self, alias)

Return all available versions for the specified alias.

Source code in kiara/data/registry/__init__.py

def get_versions_for_alias(self, alias: str) -> typing.Iterable[int]:
    """Return all available versions for the specified alias."""

    if isinstance(alias, str):
        _alias = ValueAlias.from_string(alias)
    elif not isinstance(alias, ValueAlias):
        raise TypeError(f"Invalid type for alias: {type(alias)}")
    else:
        _alias = alias

    value_slot = self.get_value_slot(alias)
    if not value_slot:
        raise Exception(f"No alias '{_alias.alias}' registered with registry.")

    return sorted(value_slot.values.keys())

DataRegistry

register_alias(self, value_or_schema, alias_name=None, callbacks=None)

Register a value slot.

A value slot is an object that holds multiple versions of values that all use the same schema.

Parameters:

Name	Type	Description	Default
value_or_schema	Union[kiara.data.values.Value, kiara.data.values.ValueSchema, str]	a value, value_id, or schema, if value, it is added to the newly created slot (after preseed, if selected)	required
preseed		whether to add an empty/non-set value to the newly created slot	required
alias_name	Optional[str]	the alias name, will be auto-created if not provided	None
callbacks	Optional[Iterable[kiara.data.registry.__init__.ValueSlotUpdateHandler]]	a list of callbacks to trigger whenever the alias is updated	None

Source code in kiara/data/registry/__init__.py

def register_alias(
    self,
    value_or_schema: typing.Union[Value, ValueSchema, str],
    # preseed: bool = False,
    alias_name: typing.Optional[str] = None,
    callbacks: typing.Optional[typing.Iterable[ValueSlotUpdateHandler]] = None,
) -> ValueSlot:
    """Register a value slot.

    A value slot is an object that holds multiple versions of values that all use the same schema.

    Arguments:
        value_or_schema: a value, value_id, or schema, if value, it is added to the newly created slot (after preseed, if selected)
        preseed: whether to add an empty/non-set value to the newly created slot
        alias_name: the alias name, will be auto-created if not provided
        callbacks: a list of callbacks to trigger whenever the alias is updated
    """

    if alias_name is None:
        alias_name = str(uuid.uuid4())

    if not alias_name:
        raise Exception("Empty alias name not allowed.")

    match = self._check_valid_alias(alias_name)
    if not match:
        raise Exception(
            f"Invalid alias '{alias_name}': only alphanumeric characters, '-', and '_' allowed in alias name."
        )

    # if the provided value_or_schema argument is a string, it must be a value_id (for now)
    if isinstance(value_or_schema, str):
        if value_or_schema not in self.value_ids:
            raise Exception(
                f"Can't register alias '{alias_name}', provided string '{value_or_schema}' not an existing value id."
            )
        _value_or_schema = self.get_value_obj(value_or_schema)
        if _value_or_schema is None:
            raise Exception(f"No value registered for: {value_or_schema}")
        value_or_schema = _value_or_schema

    if isinstance(value_or_schema, Value):
        _value_schema = value_or_schema.value_schema
        _value: typing.Optional[Value] = value_or_schema
    elif isinstance(value_or_schema, ValueSchema):
        _value_schema = value_or_schema
        _value = None
    else:
        raise TypeError(f"Invalid value type: {type(value_or_schema)}")

    if alias_name in self._get_available_aliases():
        raise Exception(
            f"Can't register alias: alias '{alias_name}' already exists."
        )
    elif alias_name in self._get_available_value_ids():
        raise Exception(
            f"Can't register alias: alias '{alias_name}' already exists as value id."
        )

    # vs = ValueSlot.from_value(id=_id, value=value_or_schema)
    vs = self._register_alias(alias_name=alias_name, value_schema=_value_schema)
    # self._value_slots[vs.id] = vs

    # if preseed:
    #     _v = Value(value_schema=_value_schema, kiara=self._kiara, registry=self)
    #     vs.add_value(_v)

    if callbacks:
        vs.register_callbacks(*callbacks)
        # self.register_callbacks(vs, *callbacks)

    if _value is not None:
        vs.add_value(_value)

    return vs

ValueSlotUpdateHandler

The call signature for callbacks that can be registered as value update handlers.