-
Notifications
You must be signed in to change notification settings - Fork 77
/
plugins.py
197 lines (151 loc) · 6.18 KB
/
plugins.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
"""
Support for plugins to sourmash via importlib.metadata entrypoints.
Plugin entry point names:
* 'sourmash.load_from' - Index class loading.
* 'sourmash.save_to' - Signature saving.
* 'sourmash.cli_script' - command-line extension.
CTB TODO:
* consider using something other than 'name' for loader fn name. Maybe __doc__?
* try implement picklist plugin?
"""
DEFAULT_LOAD_FROM_PRIORITY = 99
DEFAULT_SAVE_TO_PRIORITY = 99
import itertools
import argparse
from .logging import (debug_literal, error, notify, set_quiet)
# cover for older versions of Python that don't support selection on load
# (the 'group=' below).
from importlib.metadata import entry_points
# load 'load_from' entry points. NOTE: this executes on import of this module.
try:
_plugin_load_from = entry_points(group='sourmash.load_from')
except TypeError:
from importlib_metadata import entry_points
_plugin_load_from = entry_points(group='sourmash.load_from')
# load 'save_to' entry points as well.
_plugin_save_to = entry_points(group='sourmash.save_to')
# aaaaand CLI entry points:
_plugin_cli = entry_points(group='sourmash.cli_script')
_plugin_cli_once = False
###
def get_load_from_functions():
"Load the 'load_from' plugins and yield tuples (priority, name, fn)."
debug_literal(f"load_from plugins: {_plugin_load_from}")
# Load each plugin,
for plugin in _plugin_load_from:
try:
loader_fn = plugin.load()
except (ModuleNotFoundError, AttributeError) as e:
debug_literal(f"plugins.load_from_functions: got error loading {plugin.name}: {str(e)}")
continue
# get 'priority' if it is available
priority = getattr(loader_fn, 'priority', DEFAULT_LOAD_FROM_PRIORITY)
# retrieve name (which is specified by plugin?)
name = plugin.name
debug_literal(f"plugins.load_from_functions: got '{name}', priority={priority}")
yield priority, name, loader_fn
def get_save_to_functions():
"Load the 'save_to' plugins and yield tuples (priority, fn)."
debug_literal(f"save_to plugins: {_plugin_save_to}")
# Load each plugin,
for plugin in _plugin_save_to:
try:
save_cls = plugin.load()
except (ModuleNotFoundError, AttributeError) as e:
debug_literal(f"plugins.load_from_functions: got error loading {plugin.name}: {str(e)}")
continue
# get 'priority' if it is available
priority = getattr(save_cls, 'priority', DEFAULT_SAVE_TO_PRIORITY)
# retrieve name (which is specified by plugin?)
name = plugin.name
debug_literal(f"plugins.save_to_functions: got '{name}', priority={priority}")
yield priority, save_cls
class CommandLinePlugin:
"""
Provide some minimal common CLI functionality - -q and -d.
Subclasses should call super().__init__(parser) and super().main(args).
"""
command = None
description = None
def __init__(self, parser):
parser.add_argument(
'-q', '--quiet', action='store_true',
help='suppress non-error output'
)
parser.add_argument(
'-d', '--debug', action='store_true',
help='provide debugging output'
)
def main(self, args):
set_quiet(args.quiet, args.debug)
def get_cli_script_plugins():
global _plugin_cli_once
x = []
for plugin in _plugin_cli:
name = plugin.name
mod = plugin.module
try:
script_cls = plugin.load()
except (ModuleNotFoundError, AttributeError):
if _plugin_cli_once is False:
error(f"ERROR: cannot find or load module for cli_script plugin '{name}'")
continue
command = getattr(script_cls, 'command', None)
if command is None:
# print error message only once...
if _plugin_cli_once is False:
error(f"ERROR: no command provided by cli_script plugin '{name}' from {mod}; skipping")
else:
x.append(plugin)
_plugin_cli_once = True
return x
def get_cli_scripts_descriptions():
"Build the descriptions for command-line plugins."
for plugin in get_cli_script_plugins():
name = plugin.name
script_cls = plugin.load()
command = getattr(script_cls, 'command')
description = getattr(script_cls, 'description', "")
if description:
description = description.splitlines()[0]
if not description:
description = f"(no description provided by plugin '{name}')"
yield f"sourmash scripts {command:16s} - {description}"
def add_cli_scripts(parser):
"Configure parsing for command-line plugins."
d = {}
for plugin in get_cli_script_plugins():
name = plugin.name
script_cls = plugin.load()
usage = getattr(script_cls, 'usage', None)
description = getattr(script_cls, 'description', None)
epilog = getattr(script_cls, 'epilog', None)
formatter_class = getattr(script_cls, 'formatter_class',
argparse.HelpFormatter)
subparser = parser.add_parser(script_cls.command,
usage=usage,
description=description,
epilog=epilog,
formatter_class=formatter_class)
debug_literal(f"cls_script plugin '{name}' adding command '{script_cls.command}'")
obj = script_cls(subparser)
d[script_cls.command] = obj
return d
def list_all_plugins():
plugins = itertools.chain(_plugin_load_from,
_plugin_save_to,
_plugin_cli)
plugins = list(plugins)
if not plugins:
notify("\n(no plugins detected)\n")
notify("")
notify("the following plugins are installed:")
notify("")
notify(f"{'plugin type':<20s} {'from python module':<30s} {'v':<5s} {'entry point name':<20s}")
notify(f"{'-'*20} {'-'*30} {'-'*5} {'-'*20}")
for plugin in plugins:
name = plugin.name
mod = plugin.module
version = plugin.dist.version
group = plugin.group
notify(f"{group:<20s} {mod:<30s} {version:<5s} {name:<20s}")