Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.util.plugin_config

from pathlib import Path
from types import ModuleType
from typing import (
    Any,
    cast,
    Dict,
    Generator,
    Iterable,
    List,
    NamedTuple,
    Optional,
    Type,
    Union,
)

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)