Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/galaxy/util/plugin_config.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import collections | |
2 | |
3 import yaml | |
4 | |
5 from galaxy.util import parse_xml | |
6 from galaxy.util.submodules import import_submodules | |
7 | |
8 | |
9 PluginConfigSource = collections.namedtuple('PluginConfigSource', ['type', 'source']) | |
10 | |
11 | |
12 def plugins_dict(module, plugin_type_identifier): | |
13 """ Walk through all classes in submodules of module and find ones labelled | |
14 with specified plugin_type_identifier and throw in a dictionary to allow | |
15 constructions from plugins by these types later on. | |
16 """ | |
17 plugin_dict = {} | |
18 | |
19 for plugin_module in import_submodules(module, ordered=True): | |
20 for clazz in __plugin_classes_in_module(plugin_module): | |
21 plugin_type = getattr(clazz, plugin_type_identifier, None) | |
22 if plugin_type: | |
23 plugin_dict[plugin_type] = clazz | |
24 | |
25 return plugin_dict | |
26 | |
27 | |
28 def load_plugins(plugins_dict, plugin_source, extra_kwds=None, plugin_type_keys=('type',), dict_to_list_key=None): | |
29 if extra_kwds is None: | |
30 extra_kwds = {} | |
31 if plugin_source.type == "xml": | |
32 return __load_plugins_from_element(plugins_dict, plugin_source.source, extra_kwds) | |
33 else: | |
34 return __load_plugins_from_dicts( | |
35 plugins_dict, | |
36 plugin_source.source, | |
37 extra_kwds, | |
38 plugin_type_keys=plugin_type_keys, | |
39 dict_to_list_key=dict_to_list_key, | |
40 ) | |
41 | |
42 | |
43 def __plugin_classes_in_module(plugin_module): | |
44 for clazz in getattr(plugin_module, "__all__", []): | |
45 try: | |
46 clazz = getattr(plugin_module, clazz) | |
47 except TypeError: | |
48 clazz = clazz | |
49 yield clazz | |
50 | |
51 | |
52 def __load_plugins_from_element(plugins_dict, plugins_element, extra_kwds): | |
53 plugins = [] | |
54 | |
55 for plugin_element in plugins_element: | |
56 plugin_type = plugin_element.tag | |
57 plugin_kwds = dict(plugin_element.items()) | |
58 plugin_kwds.update(extra_kwds) | |
59 try: | |
60 plugin_klazz = plugins_dict[plugin_type] | |
61 except KeyError: | |
62 template = "Failed to find plugin of type [%s] in available plugin types %s" | |
63 message = template % (plugin_type, str(plugins_dict.keys())) | |
64 raise Exception(message) | |
65 | |
66 plugin = plugin_klazz(**plugin_kwds) | |
67 plugins.append(plugin) | |
68 | |
69 return plugins | |
70 | |
71 | |
72 def __load_plugins_from_dicts(plugins_dict, configs, extra_kwds, plugin_type_keys, dict_to_list_key): | |
73 plugins = [] | |
74 | |
75 if isinstance(configs, dict) and dict_to_list_key is not None: | |
76 configs_as_list = [] | |
77 for key, value in configs.items(): | |
78 config = value.copy() | |
79 config[dict_to_list_key] = key | |
80 configs_as_list.append(config) | |
81 else: | |
82 configs_as_list = configs | |
83 | |
84 for config in configs_as_list: | |
85 plugin_type = None | |
86 for plugin_type_key in plugin_type_keys: | |
87 if plugin_type_key in config: | |
88 plugin_type = config[plugin_type_key] | |
89 break | |
90 assert plugin_type is not None, "Could not determine plugin type for [%s]" % config | |
91 plugin_kwds = config | |
92 plugin_kwds.update(extra_kwds) | |
93 plugin = plugins_dict[plugin_type](**plugin_kwds) | |
94 plugins.append(plugin) | |
95 | |
96 return plugins | |
97 | |
98 | |
99 def plugin_source_from_path(path): | |
100 if path.endswith(".yaml") or path.endswith(".yml") or path.endswith(".yaml.sample") or path.endswith(".yml.sample"): | |
101 return PluginConfigSource('dict', __read_yaml(path)) | |
102 else: | |
103 return PluginConfigSource('xml', parse_xml(path, remove_comments=True).getroot()) | |
104 | |
105 | |
106 def plugin_source_from_dict(as_dict): | |
107 return PluginConfigSource('dict', as_dict) | |
108 | |
109 | |
110 def __read_yaml(path): | |
111 if yaml is None: | |
112 raise ImportError("Attempting to read YAML configuration file - but PyYAML dependency unavailable.") | |
113 | |
114 with open(path, "rb") as f: | |
115 return yaml.safe_load(f) |