forked from errbotio/errbot
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathspecific_plugin_manager.py
More file actions
88 lines (74 loc) · 3.69 KB
/
specific_plugin_manager.py
File metadata and controls
88 lines (74 loc) · 3.69 KB
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
import logging
import sys
import traceback
from yapsy.PluginManager import PluginManager
from yapsy.PluginFileLocator import PluginFileLocator, PluginFileAnalyzerWithInfoFile
from .utils import collect_roots
log = logging.getLogger(__name__)
class SpecificPluginLocator(PluginFileAnalyzerWithInfoFile):
"""
This is a plugin locator (kind of filter in yapsy jargon) to match a backend.
We have to go through hoops because yapsy is really aggressive at instanciating plugin.
(this would instantiate several bots, we don't want to do that).
"""
def __init__(self, name_to_find):
super().__init__('SpecificBackendLocator', 'plug')
self._name_to_find = name_to_find
def getInfosDictFromPlugin(self, dirpath, filename):
plugin_info_dict, config_parser = super().getInfosDictFromPlugin(dirpath, filename)
if plugin_info_dict['name'] != self._name_to_find:
# reject
return None, config_parser
return plugin_info_dict, config_parser
class SpecificPluginManager(PluginManager):
""" SpecificPluginManager is a customized plugin manager to enumerate plugins
and load only a specific one.
"""
def __init__(self, bot_config, category, base_class, base_search_dir, extra_search_dirs=()):
self._config = bot_config
# set a locator that gets every possible backends as a first discovery pass.
self._locator = PluginFileLocator(analyzers=[PluginFileAnalyzerWithInfoFile('SpecificLocator', 'plug')])
self._locator.disableRecursiveScan() # This is done below correctly with find_roots_with_extra
super().__init__(plugin_locator=self._locator)
self.setCategoriesFilter({category: base_class})
all_plugins_paths = collect_roots((base_search_dir, extra_search_dirs))
log.info('%s search paths %s', category, all_plugins_paths)
self.setPluginPlaces(all_plugins_paths)
for entry in all_plugins_paths:
if entry not in sys.path:
sys.path.append(entry) # so backends can relatively import their submodules
self.locatePlugins()
log.info('Found those plugings available:')
for (_, _, plug) in self.getPluginCandidates():
log.info('\t%10s (%s)' % (plug.name, plug.path + '.py'))
def instanciateElement(self, element):
""" Override the loading method to inject config
:param element: plugin class to load.
"""
log.debug("Class to load %s" % element.__name__)
return element(self._config)
def get_candidate(self, name):
""" Find the plugin by name.
:param name: The name of the plugin you are looking for.
:return: :raise Exception:
"""
for (_, _, plug) in self.getPluginCandidates():
if plug.name == name:
return plug
raise Exception("Plugin '%s' not found." % name)
def get_plugin_by_name(self, name):
# set a locator to narrow it to only one.
self._locator.setAnalyzers([SpecificPluginLocator(name)])
log.debug("Refilter the plugins...")
self.locatePlugins()
log.debug("Load the one remaining...")
plugins = self.loadPlugins()
if len(plugins) == 0:
raise Exception("Could not find the plugin '%s'." % name)
if len(plugins) != 1:
raise Exception("There are 2 plugins with the name '%s'." % name)
if plugins[0].error is not None:
reason = plugins[0].error
formatted_error = "%s:\n%s" % (reason[0], ''.join(traceback.format_tb(plugins[0].error[2])))
raise Exception('Error loading plugin %s:\nError:\n%s\n' % (name, formatted_error))
return plugins[0].plugin_object