Source code for galaxy.util.plugin_config

from xml.etree import ElementTree

    import yaml
except ImportError:
    yaml = None

from galaxy.util.submodules 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 submodules(module): # 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)