kiara.utils.class_loading

find_all_kiara_modules()

Find all KiaraModule subclasses via package entry points.

TODO

Source code in kiara/utils/class_loading.py

def find_all_kiara_modules() -> typing.Dict[str, typing.Type["KiaraModule"]]:
    """Find all [KiaraModule][kiara.module.KiaraModule] subclasses via package entry points.

    TODO
    """

    from kiara.module import KiaraModule

    modules = load_all_subclasses_for_entry_point(
        entry_point_name="kiara.modules",
        base_class=KiaraModule,  # type: ignore
        set_id_attribute="_module_type_name",
        remove_namespace_tokens=["core."],
    )
    result = {}
    # need to test this, since I couldn't add an abstract method to the KiaraModule class itself (mypy complained because it is potentially overloaded)
    for k, cls in modules.items():

        if not hasattr(cls, "process"):
            msg = f"Ignoring module class '{cls}': no 'process' method."
            if is_debug():
                log.warning(msg)
            else:
                log.debug(msg)
            continue

        # TODO: check signature of process method

        if k.startswith("_"):
            tokens = k.split(".")
            if len(tokens) == 1:
                k = k[1:]
            else:
                k = ".".join(tokens[1:])

        result[k] = cls
    return result

find_all_metadata_models()

Find all KiaraModule subclasses via package entry points.

TODO

Source code in kiara/utils/class_loading.py

def find_all_metadata_models() -> typing.Dict[str, typing.Type["MetadataModel"]]:
    """Find all [KiaraModule][kiara.module.KiaraModule] subclasses via package entry points.

    TODO
    """

    return load_all_subclasses_for_entry_point(
        entry_point_name="kiara.metadata_models",
        base_class=MetadataModel,
        set_id_attribute="_metadata_key",
        remove_namespace_tokens=["core."],
    )

find_all_value_types()

Find all KiaraModule subclasses via package entry points.

TODO

Source code in kiara/utils/class_loading.py

def find_all_value_types() -> typing.Dict[str, typing.Type["ValueType"]]:
    """Find all [KiaraModule][kiara.module.KiaraModule] subclasses via package entry points.

    TODO
    """

    all_value_types = load_all_subclasses_for_entry_point(
        entry_point_name="kiara.value_types",
        base_class=ValueType,  # type: ignore
        set_id_attribute="_value_type_name",
        remove_namespace_tokens=True,
    )

    invalid = [x for x in all_value_types.keys() if "." in x]
    if invalid:
        raise Exception(
            f"Invalid value type name(s), type names can't contain '.': {', '.join(invalid)}"
        )

    return all_value_types

find_subclasses_under(base_class, module, prefix='', remove_namespace_tokens=None, module_name_func=None)

Find all (non-abstract) subclasses of a base class that live under a module (recursively).

Parameters:

Name	Type	Description	Default
base_class	Type[~SUBCLASS_TYPE]	the parent class	required
module	Union[str, module]	the module to search	required
prefix	Optional[str]	a string to use as a result items namespace prefix, defaults to an empty string, use 'None' to indicate the module path should be used	''
remove_namespace_tokens	Optional[Iterable[str]]	a list of strings to remove from module names when autogenerating subclass ids, and prefix is None	None

Returns:

Type	Description
Mapping[str, Type[~SUBCLASS_TYPE]]	a map containing the (fully namespaced) id of the subclass as key, and the actual class object as value

Source code in kiara/utils/class_loading.py

def find_subclasses_under(
    base_class: typing.Type[SUBCLASS_TYPE],
    module: typing.Union[str, ModuleType],
    prefix: typing.Optional[str] = "",
    remove_namespace_tokens: typing.Optional[typing.Iterable[str]] = None,
    module_name_func: typing.Callable = None,
) -> typing.Mapping[str, typing.Type[SUBCLASS_TYPE]]:
    """Find all (non-abstract) subclasses of a base class that live under a module (recursively).

    Arguments:
        base_class: the parent class
        module: the module to search
        prefix: a string to use as a result items namespace prefix, defaults to an empty string, use 'None' to indicate the module path should be used
        remove_namespace_tokens: a list of strings to remove from module names when autogenerating subclass ids, and prefix is None

    Returns:
        a map containing the (fully namespaced) id of the subclass as key, and the actual class object as value
    """

    if hasattr(sys, "frozen"):
        raise NotImplementedError("Pyinstaller bundling not supported yet.")

    if isinstance(module, str):
        module = importlib.import_module(module)

    _import_modules_recursively(module)

    subclasses: typing.Iterable[typing.Type[SUBCLASS_TYPE]] = _get_all_subclasses(
        base_class
    )

    result = {}
    for sc in subclasses:

        if not sc.__module__.startswith(module.__name__):
            continue

        if inspect.isabstract(sc):
            if is_debug():
                # import traceback
                # traceback.print_stack()
                log.warning(f"Ignoring abstract subclass: {sc}")
            else:
                log.debug(f"Ignoring abstract subclass: {sc}")
            continue

        if module_name_func is None:
            module_name_func = _get_subclass_name
        name = module_name_func(sc)
        path = sc.__module__[len(module.__name__) + 1 :]  # noqa

        if path:
            full_name = f"{path}.{name}"
        else:
            full_name = name

        if prefix is None:
            prefix = module.__name__ + "."
            if remove_namespace_tokens:
                for rnt in remove_namespace_tokens:
                    if prefix.startswith(rnt):
                        prefix = prefix[0 : -len(rnt)]  # noqa

        if prefix:
            full_name = f"{prefix}.{full_name}"

        result[full_name] = sc

    return result

load_all_subclasses_for_entry_point(entry_point_name, base_class, set_id_attribute=None, remove_namespace_tokens=None)

Find all subclasses of a base class via package entry points.

Parameters:

Name	Type	Description	Default
entry_point_name	str	the entry point name to query entries for	required
base_class	Type[~SUBCLASS_TYPE]	the base class to look for	required
set_id_attribute	Optional[str]	whether to set the entry point id as attribute to the class, if None, no id attribute will be set, if a string, the attribute with that name will be set	None
remove_namespace_tokens	Union[Iterable[str], bool]	a list of strings to remove from module names when autogenerating subclass ids, and prefix is None, or a boolean in which case all or none namespaces will be removed	None

TODO

Source code in kiara/utils/class_loading.py

def load_all_subclasses_for_entry_point(
    entry_point_name: str,
    base_class: typing.Type[SUBCLASS_TYPE],
    set_id_attribute: typing.Union[None, str] = None,
    remove_namespace_tokens: typing.Union[typing.Iterable[str], bool, None] = None,
) -> typing.Dict[str, typing.Type[SUBCLASS_TYPE]]:
    """Find all subclasses of a base class via package entry points.

    Arguments:
        entry_point_name: the entry point name to query entries for
        base_class: the base class to look for
        set_id_attribute: whether to set the entry point id as attribute to the class, if None, no id attribute will be set, if a string, the attribute with that name will be set
        remove_namespace_tokens: a list of strings to remove from module names when autogenerating subclass ids, and prefix is None, or a boolean in which case all or none namespaces will be removed

    TODO
    """

    log2 = logging.getLogger("stevedore")
    out_hdlr = logging.StreamHandler(sys.stdout)
    out_hdlr.setFormatter(
        logging.Formatter(f"{entry_point_name} plugin search error -> %(message)s")
    )
    out_hdlr.setLevel(logging.INFO)
    log2.addHandler(out_hdlr)
    log2.setLevel(logging.INFO)

    log.debug(f"Finding {entry_point_name} items from search paths...")

    mgr = ExtensionManager(
        namespace=entry_point_name,
        invoke_on_load=False,
        propagate_map_exceptions=True,
    )

    result_entrypoints: typing.Dict[str, typing.Type] = {}
    result_dynamic: typing.Dict[str, typing.Type] = {}
    for plugin in mgr:
        name = plugin.name

        if isinstance(plugin.plugin, type) and issubclass(plugin.plugin, base_class):
            ep = plugin.entry_point
            module_cls = ep.load()

            if set_id_attribute:
                if hasattr(module_cls, set_id_attribute):
                    if not getattr(module_cls, set_id_attribute) == name:
                        log.warning(
                            f"Item id mismatch for type {entry_point_name}: {getattr(module_cls, set_id_attribute)} != {name}, entry point key takes precedence: {name})"
                        )
                        setattr(module_cls, set_id_attribute, name)

                else:
                    setattr(module_cls, set_id_attribute, name)
            result_entrypoints[name] = module_cls
        elif (
            isinstance(plugin.plugin, tuple)
            and len(plugin.plugin) >= 1
            and callable(plugin.plugin[0])
        ) or callable(plugin.plugin):
            modules = _callable_wrapper(plugin.plugin)

            for k, v in modules.items():
                _name = f"{name}.{k}"
                if _name in result_dynamic.keys():
                    raise Exception(
                        f"Duplicate item name for type {entry_point_name}: {_name}"
                    )
                result_dynamic[_name] = v

        else:
            raise Exception(
                f"Can't load subclasses for entry point {entry_point_name} and base class {base_class}: invalid plugin type {type(plugin.plugin)}"
            )

    for k, v in result_dynamic.items():
        if k in result_entrypoints.keys():
            raise Exception(f"Duplicate item name for type {entry_point_name}: {k}")
        result_entrypoints[k] = v

    result: typing.Dict[str, typing.Type[SUBCLASS_TYPE]] = {}
    for k, v in result_entrypoints.items():

        if remove_namespace_tokens:
            if remove_namespace_tokens is True:
                k = k.split(".")[-1]
            elif isinstance(remove_namespace_tokens, typing.Iterable):
                for rnt in remove_namespace_tokens:
                    if k.startswith(rnt):
                        k = k[len(rnt) :]  # noqa

        if k in result.keys():
            msg = ""
            if set_id_attribute:
                msg = f" Check whether '{v.__name__}' is missing the '{set_id_attribute}' class attribute (in case this is a sub-class), or it's '{k}' value is also set in another class?"
            raise Exception(
                f"Duplicate item name for base class {base_class}: {k}.{msg}"
            )
        result[k] = v

    return result