Source code for galaxy.util.plugin_config

from pathlib import Path
from types import ModuleType
from typing import (

import yaml

from galaxy.util import parse_xml
from galaxy.util.submodules import import_submodules

PathT = Union[str, Path]
PluginDictConfigT = Dict[str, Any]
PluginConfigsT = Union[PluginDictConfigT, List[PluginDictConfigT]]

[docs]class PluginConfigSource(NamedTuple): type: str source: Any
[docs]def plugins_dict(module: ModuleType, plugin_type_identifier: str) -> Dict[str, Type]: """Walk through all classes in submodules of module and find ones labelled with specified plugin_type_identifier and throw in a dictionary to allow constructions from plugins by these types later on. """ plugin_dict = {} for plugin_module in import_submodules(module, ordered=True): for clazz in __plugin_classes_in_module(plugin_module): plugin_type = getattr(clazz, plugin_type_identifier, None) if plugin_type: plugin_dict[plugin_type] = clazz return plugin_dict
[docs]def load_plugins( plugins_dict: Dict[str, Type], plugin_source: PluginConfigSource, extra_kwds: Optional[Dict[str, Any]] = None, plugin_type_keys: Iterable[str] = ("type",), dict_to_list_key: Optional[str] = None, ) -> List[Any]: if extra_kwds is None: extra_kwds = {} if plugin_source.type == "xml": return __load_plugins_from_element(plugins_dict, plugin_source.source, extra_kwds) else: return __load_plugins_from_dicts( plugins_dict, plugin_source.source, extra_kwds, plugin_type_keys=plugin_type_keys, dict_to_list_key=dict_to_list_key, )
def __plugin_classes_in_module(plugin_module: ModuleType) -> Generator[Type, None, None]: for clazz in getattr(plugin_module, "__all__", []): try: clazz = getattr(plugin_module, clazz) except TypeError: clazz = clazz yield clazz def __load_plugins_from_element( plugins_dict: Dict[str, Type], plugins_element, extra_kwds: Dict[str, Any] ) -> List[Any]: plugins = [] for plugin_element in plugins_element: plugin_type = plugin_element.tag plugin_kwds = dict(plugin_element.items()) plugin_kwds.update(extra_kwds) try: plugin_klazz = plugins_dict[plugin_type] except KeyError: template = "Failed to find plugin of type [%s] in available plugin types %s" message = template % (plugin_type, str(plugins_dict.keys())) raise Exception(message) plugin = plugin_klazz(**plugin_kwds) plugins.append(plugin) return plugins def __load_plugins_from_dicts( plugins_dict: Dict[str, Type], configs: PluginConfigsT, extra_kwds: Dict[str, Any], plugin_type_keys: Iterable[str], dict_to_list_key: Optional[str], ) -> List[Any]: plugins = [] configs_as_list: List[PluginDictConfigT] if isinstance(configs, dict) and dict_to_list_key is not None: configs_as_list = [] for key, value in configs.items(): config = value.copy() config[dict_to_list_key] = key configs_as_list.append(config) else: configs_as_list = cast(List[PluginDictConfigT], configs) for config in configs_as_list: plugin_type = None for plugin_type_key in plugin_type_keys: if plugin_type_key in config: plugin_type = config[plugin_type_key] break assert plugin_type is not None, f"Could not determine plugin type for [{config}]" plugin_kwds = config if extra_kwds: plugin_kwds = plugin_kwds.copy() plugin_kwds.update(extra_kwds) plugin = plugins_dict[plugin_type](**plugin_kwds) plugins.append(plugin) return plugins
[docs]def plugin_source_from_path(path: PathT) -> PluginConfigSource: filename = str(path) if ( filename.endswith(".yaml") or filename.endswith(".yml") or filename.endswith(".yaml.sample") or filename.endswith(".yml.sample") ): return PluginConfigSource("dict", __read_yaml(path)) else: return PluginConfigSource("xml", parse_xml(path, remove_comments=True).getroot())
[docs]def plugin_source_from_dict(as_dict: PluginConfigsT) -> PluginConfigSource: return PluginConfigSource("dict", as_dict)
def __read_yaml(path: PathT): if yaml is None: raise ImportError("Attempting to read YAML configuration file - but PyYAML dependency unavailable.") with open(path, "rb") as f: return yaml.safe_load(f)