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 xml.etree import ElementTree

    import yaml
except ImportError:
    yaml = None

from galaxy.util.submodules import import_submodules

[docs]def plugins_dict(module, plugin_type_identifier): """ 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): # FIXME: this is not how one is suppose to use __all__ why did you do # this past John? for clazz in getattr(plugin_module, "__all__", []): try: clazz = getattr(plugin_module, clazz) except TypeError: clazz = clazz 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, plugin_source, extra_kwds={}): source_type, source = plugin_source if source_type == "xml": return __load_plugins_from_element(plugins_dict, source, extra_kwds) else: return __load_plugins_from_dicts(plugins_dict, source, extra_kwds)
def __load_plugins_from_element(plugins_dict, plugins_element, extra_kwds): 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, configs, extra_kwds): plugins = [] for config in configs: plugin_type = config["type"] plugin_kwds = config plugin_kwds.update(extra_kwds) plugin = plugins_dict[plugin_type](**plugin_kwds) plugins.append(plugin) return plugins
[docs]def plugin_source_from_path(path): if path.endswith(".yaml") or path.endswith(".yml") or path.endswith(".yaml.sample") or path.endswith(".yml.sample"): return ('dict', __read_yaml(path)) else: return ('xml', ElementTree.parse(path).getroot())
def __read_yaml(path): 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)