PluginManager.initialise() - Code Metrics - Inspection of "Merge branch 'master' of https://github.com/useblo..." - useblocks/groundwork - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 754800...aae51d )

by Daniel

created 2016-09-22 06:15 UTC

PluginManager.initialise() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	13
c	1
b	0
f	1
dl	0
loc	40
rs	2.7716

How to fix Complexity

"""
The pluginmanager module cares about the management of plugin status and their changes between status.

There are two manager classes for managing plugin related objects.

 * PluginManager: Cares about initialised Plugins, which can be activated and deactivated.
 * PluginClassManager: Cares about plugin classes, which are used to create plugins.

A plugin class can be reused for several plugins. The only thing to care about is the naming of a plugin.
This plugin name must be unique inside a groundwork app and can be set during plugin initialisation/activation.
"""

from pkg_resources import iter_entry_points
import logging
import inspect

from groundwork.patterns.gw_base_pattern import GwBasePattern
from groundwork.exceptions import PluginNotActivatableException, PluginNotInitialisableException, \
    PluginRegistrationException, PluginNotDeactivatableException


class PluginManager:
    """
    PluginManager for searching, initialising, activating and deactivating groundwork plugins.
    """

    def __init__(self, app, strict=False):
        """
        Initialises the plugin manager.

        Additional plugins can be registered by adding their class via the plugins argument.

        :param app: groundwork application object
        :param strict: If True, problems during plugin registration/initialisation or activation will throw an exception
        """
        self._log = logging.getLogger(__name__)
        self._strict = strict
        self._app = app
        self._plugins = {}

        #: Instance of :class:`~groundwork.pluginmanager.PluginClassManager`.
        #: Handles the registration of plugin classes, which can be used to create new plugins during runtime.
        self.classes = PluginClassManager(self._app, self._strict)

    def initialise_by_names(self, plugins=None):
        """
        Initialises given plugins, but does not activate them.

        This is needed to import and configure libraries, which are imported by used patterns, like GwFlask.

        After this action, all needed python modules are imported and configured.
        Also the groundwork application object is ready and contains functions and objects, which were added
        by patterns, like app.commands from GwCommandsPattern.

        The class of a given plugin must already be registered in the :class:`.PluginClassManager`.

        :param plugins: List of plugin names
        :type plugins: list of strings
        """

        if plugins is None:
            plugins = []

        self._log.debug("Plugins Initialisation started")
        if not isinstance(plugins, list):
            raise AttributeError("plugins must be a list, not %s" % type(plugins))

        self._log.debug("Plugins to initialise: %s" % ", ".join(plugins))
        plugin_initialised = []
        for plugin_name in plugins:
            if not isinstance(plugin_name, str):
                raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))

            plugin_class = self.classes.get(plugin_name)
            self.initialise(plugin_class.clazz, plugin_name)

        self._log.info("Plugins initialised: %s" % ", ".join(plugin_initialised))

    def initialise(self, clazz, name=None):
        if clazz is not None and issubclass(clazz, GwBasePattern):
            if name is None:
                name = clazz.__name__
            try:
                # Plugin Initialisation
                plugin_instance = clazz(app=self._app, name=name)
                # Let's be sure the correct name was set. Even if the init_routine of
                # the plugin does not support the name parameter.
                if plugin_instance.name != name:
                    plugin_instance.name = name
            except Exception as e:
                self._log.warning("Plugin class %s could not be initialised" % clazz.__name__)
                if self._strict:
                    raise PluginNotInitialisableException("Plugin class %s could not be initialised" % clazz.__name__) \
                        from e

            # Let's be sure, that GwBasePattern got called
            if not hasattr(plugin_instance, "_plugin_base_initialised") \
                    or plugin_instance._plugin_base_initialised is not True:
                self._log.error("GwBasePattern.__init__() was not called during initialisation. "
                                "Please add 'super(*args, **kwargs).__init__()' to the top of all involved "
                                "plugin/pattern init routines."
                                "Activate logging debug-output to see all involved classes.")
                for mro_class in clazz.__mro__:
                    self._log.debug(mro_class)
                raise Exception("GwBasePattern.__init()__ was not called during initialisation.")
            self._register_initialisation(plugin_instance)
            self._log.debug("Plugin %s initialised" % name)
            return plugin_instance

        if clazz is None:
            self._log.warn("Plugin class %s not found" % clazz.__name__)
            return None

        if not issubclass(clazz, GwBasePattern):
            self._log.warn("Can not load %s. Plugin is not based on groundwork.Plugin." % clazz.__name__)
            if self._strict:
                raise Exception("Can not load %s. Plugin is not based on groundwork.Plugin." % clazz.__name__)
        return None

    def _register_initialisation(self, plugin_instance):
        """
        Internal functions to perform registration actions after plugin load was successful.
        """
        self._plugins[plugin_instance.name] = plugin_instance

    def activate(self, plugins=[]):
        """
        Activates given plugins.

        This calls mainly plugin.activate() and plugins register needed resources like commands, signals or
        documents.

        If given plugins have not been initialised, this is also done via :func:`_load`.

        :param plugins: List of plugin names
        :type plugins: list of strings
        """
        self._log.debug("Plugins Activation started")

        if not isinstance(plugins, list):
            raise AttributeError("plugins must be a list, not %s" % type(plugins))

        self._log.debug("Plugins to activate: %s" % ", ".join(plugins))

        plugins_activated = []
        for plugin_name in plugins:
            if not isinstance(plugin_name, str):
                raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))


            if plugin_name not in self._plugins.keys() and plugin_name in self.classes._classes.keys():
                self._log.debug("Initialisation needed before activation.")
                try:
                    self.initialise_by_names([plugin_name])
                except Exception as e:
                    self._log.error("Couldn't initialise plugin %s" % plugin_name)
                    if self._strict:
                        raise Exception("Couldn't initialise plugin %s" % plugin_name) from e
                    else:
                        continue
            if plugin_name in self._plugins.keys():
                self._log.debug("Activating plugin %s" % plugin_name)
                if not self._plugins[plugin_name].active:
                    try:
                        self._plugins[plugin_name].activate()
                    except Exception as e:
                        raise PluginNotActivatableException("Plugin %s could not be activated" % plugin_name) from e
                    else:
                        self._log.debug("Plugin %s activated" % plugin_name)
                        plugins_activated.append(plugin_name)
                else:
                    self._log.warning("Plugin %s got already activated." % plugin_name)
                    if self._strict:
                        raise PluginNotInitialisableException()

        self._log.info("Plugins activated: %s" % ", ".join(plugins_activated))

    def deactivate(self, plugins=[]):
        """
        Deactivates given plugins.

        A given plugin must be activated, otherwise it is ignored and no action takes place (no signals are fired,
        no deactivate functions are called.)

        A deactivated plugin is still loaded and initialised and can be reactivated by calling :func:`activate` again.
        It is also still registered in the :class:`.PluginManager` and can be requested via :func:`get`.

        :param plugins: List of plugin names
        :type plugins: list of strings
        """
        self._log.debug("Plugins Deactivation started")


        if not isinstance(plugins, list):
            raise AttributeError("plugins must be a list, not %s" % type(plugins))

        self._log.debug("Plugins to deactivate: %s" % ", ".join(plugins))

        plugins_deactivated = []
        for plugin_name in plugins:
            if not isinstance(plugin_name, str):
                raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))

            if plugin_name not in self._plugins.keys():
                self._log.info("Unknown activated plugin %s" % plugin_name)
                continue
            else:
                self._log.debug("Deactivating plugin %s" % plugin_name)
                if not self._plugins[plugin_name].active:
                    self._log.warning("Plugin %s seems to be already deactivated" % plugin_name)
                else:
                    try:
                        self._plugins[plugin_name].deactivate()
                    except Exception as e:
                        raise PluginNotDeactivatableException("Plugin %s could not be deactivated" % plugin_name) from e
                    else:
                        self._log.debug("Plugin %s deactivated" % plugin_name)
                        plugins_deactivated.append(plugin_name)

        self._log.info("Plugins deactivated: %s" % ", ".join(plugins_deactivated))

    def get(self, name=None):
        """
        Returns the plugin object with the given name.
        Or if a name is not given, the complete plugin dictionary is returned.

        :param name: Name of a plugin
        :return: None, single plugin or dictionary of plugins
        """
        if name is None:
            return self._plugins
        else:
            if name not in self._plugins.keys():
                return None
            else:
                return self._plugins[name]

    def exist(self, name):
        """
        Returns True if plugin exists.
        :param name: plugin name
        :return: boolean
        """
        if name in self._plugins.keys():
            return True
        return False

    def is_active(self, name):
        """
        Returns True if plugin exists and is active.
        If plugin does not exist, it returns None

        :param name: plugin name
        :return: boolean or None
        """
        if name in self._plugins.keys():
            return self._plugins["name"].active
        return None


class PluginClassManager:
    """
    Manages the plugin classes, which can be used to initialise and activate new plugins.

    Loads all plugin classes from entry_point "groundwork.plugin" automatically during own initialisation.
    Provides functions to register new plugin classes during runtime.
    """

    def __init__(self, app, strict=False):
        self._log = logging.getLogger(__name__)
        self._app = app
        self._strict = strict
        self._classes = {}
        self._get_plugins_by_entry_points()

    def _get_plugins_by_entry_points(self):
        """
        Registers plugin classes, which are in sys.path and have an entry_point called 'groundwork.plugin'.
        :return: dict of plugin classes
        """
        # Let's find and register every plugin, which is in sys.path and has defined a entry_point 'groundwork.plugin'
        # in it's setup.py
        entry_points = []
        classes = {}
        for entry_point in iter_entry_points(group='groundwork.plugin', name=None):
            entry_points.append(entry_point)

        for entry_point in entry_points:
            try:
                entry_point_object = entry_point.load()
            except Exception as e:
                # We should not throw an exception now, because a package/entry_point can be outdated, using an old
                # api from groundwork, tries to import unavailable packages, what ever...
                # We just do not make it available. That's all we can do.
                self._log.warning("Couldn't load entry_point %s. Reason: %s" % (entry_point.name, e))
                continue

            if not issubclass(entry_point_object, GwBasePattern):
                self._log.warning("entry_point  %s is not a subclass of groundworkPlugin" % entry_point.name)
                continue
            plugin_name = entry_point_object.__name__

            plugin_class = self.register_class(entry_point_object, plugin_name,
                                               entrypoint_name=entry_point.name,
                                               distribution_path=entry_point.dist.location,
                                               distribution_key=entry_point.dist.key,
                                               distribution_version=entry_point.dist.version)

            classes[plugin_name] = plugin_class
            # classes[plugin_name] = {
            #     "name": plugin_name,
            #     "entry_point": entry_point.name,
            #     "path": entry_point.dist.location,
            #     "class": entry_point_object,
            #     "distribution": {
            #         "key": entry_point.dist.key,
            #         "version": entry_point.dist.version
            #     },
            # }
            self._log.debug("Found plugin: %s at entry_point %s of package %s (%s)" % (plugin_name, entry_point.name,
                                                                                       entry_point.dist.key,
                                                                                       entry_point.dist.version))
        return classes

    def register(self, classes=[]):
        """
        Registers new plugins.

        The registration only creates a new entry for a plugin inside the _classes dictionary.
        It does not activate or even initialise the plugin.

        A plugin must be a class, which inherits directly or indirectly from GwBasePattern.

        :param classes: List of plugin classes
        :type classes: list
        """
        if not isinstance(classes, list):
            raise AttributeError("plugins must be a list, not %s." % type(classes))

        plugin_registered = []

        for plugin_class in classes:
            plugin_name = plugin_class.__name__
            self.register_class(plugin_class, plugin_name)
            self._log.debug("Plugin %s registered" % plugin_name)
            plugin_registered.append(plugin_name)

        self._log.info("Plugins registered: %s" % ", ".join(plugin_registered))

    def register_class(self, clazz, name=None, entrypoint_name=None, distribution_path=None,
                       distribution_key=None, distribution_version=None):

        if name is None:
            name = clazz.__name__

        if not inspect.isclass(clazz) or not issubclass(clazz, GwBasePattern):
            self._log.error("Given plugin is not a subclass of groundworkPlugin.")
            if self._strict:
                raise AttributeError("Given plugin is not a subclass of groundworkPlugin.")

        if isinstance(clazz, GwBasePattern):
            self._log.error("Given plugin %s is already initialised. Please provide a class not an instance.")
            if self._strict:
                raise Exception("Given plugin %s is already initialised. Please provide a class not an instance.")

        if name in self._classes.keys():
            self._log.warning("Plugin %s already registered" % name)
            if self._strict:
                raise PluginRegistrationException("Plugin %s already registered" % name)

        self._classes[name] = PluginClass(name, clazz, entrypoint_name, distribution_path,
                                          distribution_key, distribution_version)

        return self._classes[name]

        # self._classes[name] = {
        #     "name": plugin_name,
        #     "entry_point": None,
        #     "path": None,
        #     "class": plugin,
        #     "distribution": None,
        # }

    def get(self, name=None):
        """
        Returns the plugin class object with the given name.
        Or if a name is not given, the complete plugin dictionary is returned.

        :param name: Name of a plugin
        :return: None, single plugin or dictionary of plugins
        """
        if name is None:
            return self._classes
        else:
            if name not in self._classes.keys():
                return None
            else:
                return self._classes[name]

    def exist(self, name):
        """
        Returns True if plugin class exists.
        :param name: plugin name
        :return: boolean
        """
        if name in self._classes.keys():
            return True
        return False


class PluginClass:
    def __init__(self, name, clazz, entrypoint_name, distribution_path, distribution_key, distribution_version):
        self.name = name
        self.entrypoint_name = entrypoint_name
        self.clazz = clazz
        self.distribution = {
            "path": distribution_path,
            "key": distribution_key,
            "version": distribution_version
        }


1		"""
2		The pluginmanager module cares about the management of plugin status and their changes between status.
3
4		There are two manager classes for managing plugin related objects.
5
6		* PluginManager: Cares about initialised Plugins, which can be activated and deactivated.
7		* PluginClassManager: Cares about plugin classes, which are used to create plugins.
8
9		A plugin class can be reused for several plugins. The only thing to care about is the naming of a plugin.
10		This plugin name must be unique inside a groundwork app and can be set during plugin initialisation/activation.
11		"""
12
13		from pkg_resources import iter_entry_points
14		import logging
15		import inspect
16
17		from groundwork.patterns.gw_base_pattern import GwBasePattern
18		from groundwork.exceptions import PluginNotActivatableException, PluginNotInitialisableException, \
19		PluginRegistrationException, PluginNotDeactivatableException
20
21
22		class PluginManager:
23		"""
24		PluginManager for searching, initialising, activating and deactivating groundwork plugins.
25		"""
26
27		def __init__(self, app, strict=False):
28		"""
29		Initialises the plugin manager.
30
31		Additional plugins can be registered by adding their class via the plugins argument.
32
33		:param app: groundwork application object
34		:param strict: If True, problems during plugin registration/initialisation or activation will throw an exception
35		"""
36		self._log = logging.getLogger(__name__)
37		self._strict = strict
38		self._app = app
39		self._plugins = {}
40
41		#: Instance of :class:`~groundwork.pluginmanager.PluginClassManager`.
42		#: Handles the registration of plugin classes, which can be used to create new plugins during runtime.
43		self.classes = PluginClassManager(self._app, self._strict)
44
45		def initialise_by_names(self, plugins=None):
46		"""
47		Initialises given plugins, but does not activate them.
48
49		This is needed to import and configure libraries, which are imported by used patterns, like GwFlask.
50
51		After this action, all needed python modules are imported and configured.
52		Also the groundwork application object is ready and contains functions and objects, which were added
53		by patterns, like app.commands from GwCommandsPattern.
54
55		The class of a given plugin must already be registered in the :class:`.PluginClassManager`.
56
57		:param plugins: List of plugin names
58		:type plugins: list of strings
59		"""
60
61		if plugins is None:
62		plugins = []
63
64		self._log.debug("Plugins Initialisation started")
65		if not isinstance(plugins, list):
66		raise AttributeError("plugins must be a list, not %s" % type(plugins))
67
68		self._log.debug("Plugins to initialise: %s" % ", ".join(plugins))
69		plugin_initialised = []
70		for plugin_name in plugins:
71		if not isinstance(plugin_name, str):
72		raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))
73
74		plugin_class = self.classes.get(plugin_name)
75		self.initialise(plugin_class.clazz, plugin_name)
76
77		self._log.info("Plugins initialised: %s" % ", ".join(plugin_initialised))
78
79		def initialise(self, clazz, name=None):
80		if clazz is not None and issubclass(clazz, GwBasePattern):
81		if name is None:
82		name = clazz.__name__
83		try:
84		# Plugin Initialisation
85		plugin_instance = clazz(app=self._app, name=name)
86		# Let's be sure the correct name was set. Even if the init_routine of
87		# the plugin does not support the name parameter.
88		if plugin_instance.name != name:
89		plugin_instance.name = name
90		except Exception as e:
91		self._log.warning("Plugin class %s could not be initialised" % clazz.__name__)
92		if self._strict:
93		raise PluginNotInitialisableException("Plugin class %s could not be initialised" % clazz.__name__) \
94		from e
95
96		# Let's be sure, that GwBasePattern got called
97		if not hasattr(plugin_instance, "_plugin_base_initialised") \
98		or plugin_instance._plugin_base_initialised is not True:
99		self._log.error("GwBasePattern.__init__() was not called during initialisation. "
100		"Please add 'super(args, *kwargs).__init__()' to the top of all involved "
101		"plugin/pattern init routines."
102		"Activate logging debug-output to see all involved classes.")
103		for mro_class in clazz.__mro__:
104		self._log.debug(mro_class)
105		raise Exception("GwBasePattern.__init()__ was not called during initialisation.")
106		self._register_initialisation(plugin_instance)
107		self._log.debug("Plugin %s initialised" % name)
108		return plugin_instance
109
110		if clazz is None:
111		self._log.warn("Plugin class %s not found" % clazz.__name__)
112		return None
113
114		if not issubclass(clazz, GwBasePattern):
115		self._log.warn("Can not load %s. Plugin is not based on groundwork.Plugin." % clazz.__name__)
116		if self._strict:
117		raise Exception("Can not load %s. Plugin is not based on groundwork.Plugin." % clazz.__name__)
118		return None
119
120		def _register_initialisation(self, plugin_instance):
121		"""
122		Internal functions to perform registration actions after plugin load was successful.
123		"""
124		self._plugins[plugin_instance.name] = plugin_instance
125
126		def activate(self, plugins=[]):
127		"""
128		Activates given plugins.
129
130		This calls mainly plugin.activate() and plugins register needed resources like commands, signals or
131		documents.
132
133		If given plugins have not been initialised, this is also done via :func:`_load`.
134
135		:param plugins: List of plugin names
136		:type plugins: list of strings
137		"""
138		self._log.debug("Plugins Activation started")
139
140		if not isinstance(plugins, list):
141		raise AttributeError("plugins must be a list, not %s" % type(plugins))
142
143		self._log.debug("Plugins to activate: %s" % ", ".join(plugins))
144
145		plugins_activated = []
146		for plugin_name in plugins:
147		if not isinstance(plugin_name, str):
148	View Code Duplication	raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))
		0 ignored issues – show Duplication introduced 2016-08-10 13:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
149
150		if plugin_name not in self._plugins.keys() and plugin_name in self.classes._classes.keys():
151		self._log.debug("Initialisation needed before activation.")
152		try:
153		self.initialise_by_names([plugin_name])
154		except Exception as e:
155		self._log.error("Couldn't initialise plugin %s" % plugin_name)
156		if self._strict:
157		raise Exception("Couldn't initialise plugin %s" % plugin_name) from e
158		else:
159		continue
160		if plugin_name in self._plugins.keys():
161		self._log.debug("Activating plugin %s" % plugin_name)
162		if not self._plugins[plugin_name].active:
163		try:
164		self._plugins[plugin_name].activate()
165		except Exception as e:
166		raise PluginNotActivatableException("Plugin %s could not be activated" % plugin_name) from e
167		else:
168		self._log.debug("Plugin %s activated" % plugin_name)
169		plugins_activated.append(plugin_name)
170		else:
171		self._log.warning("Plugin %s got already activated." % plugin_name)
172		if self._strict:
173		raise PluginNotInitialisableException()
174
175		self._log.info("Plugins activated: %s" % ", ".join(plugins_activated))
176
177		def deactivate(self, plugins=[]):
178		"""
179		Deactivates given plugins.
180
181		A given plugin must be activated, otherwise it is ignored and no action takes place (no signals are fired,
182		no deactivate functions are called.)
183
184		A deactivated plugin is still loaded and initialised and can be reactivated by calling :func:`activate` again.
185		It is also still registered in the :class:`.PluginManager` and can be requested via :func:`get`.
186
187		:param plugins: List of plugin names
188		:type plugins: list of strings
189		"""
190	View Code Duplication	self._log.debug("Plugins Deactivation started")
		0 ignored issues – show Duplication introduced 2016-08-10 13:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
191
192		if not isinstance(plugins, list):
193		raise AttributeError("plugins must be a list, not %s" % type(plugins))
194
195		self._log.debug("Plugins to deactivate: %s" % ", ".join(plugins))
196
197		plugins_deactivated = []
198		for plugin_name in plugins:
199		if not isinstance(plugin_name, str):
200		raise AttributeError("plugin name must be a str, not %s" % type(plugin_name))
201
202		if plugin_name not in self._plugins.keys():
203		self._log.info("Unknown activated plugin %s" % plugin_name)
204		continue
205		else:
206		self._log.debug("Deactivating plugin %s" % plugin_name)
207		if not self._plugins[plugin_name].active:
208		self._log.warning("Plugin %s seems to be already deactivated" % plugin_name)
209		else:
210		try:
211		self._plugins[plugin_name].deactivate()
212		except Exception as e:
213		raise PluginNotDeactivatableException("Plugin %s could not be deactivated" % plugin_name) from e
214		else:
215		self._log.debug("Plugin %s deactivated" % plugin_name)
216		plugins_deactivated.append(plugin_name)
217
218		self._log.info("Plugins deactivated: %s" % ", ".join(plugins_deactivated))
219
220		def get(self, name=None):
221		"""
222		Returns the plugin object with the given name.
223		Or if a name is not given, the complete plugin dictionary is returned.
224
225		:param name: Name of a plugin
226		:return: None, single plugin or dictionary of plugins
227		"""
228		if name is None:
229		return self._plugins
230		else:
231		if name not in self._plugins.keys():
232		return None
233		else:
234		return self._plugins[name]
235
236		def exist(self, name):
237		"""
238		Returns True if plugin exists.
239		:param name: plugin name
240		:return: boolean
241		"""
242		if name in self._plugins.keys():
243		return True
244		return False
245
246		def is_active(self, name):
247		"""
248		Returns True if plugin exists and is active.
249		If plugin does not exist, it returns None
250
251		:param name: plugin name
252		:return: boolean or None
253		"""
254		if name in self._plugins.keys():
255		return self._plugins["name"].active
256		return None
257
258
259		class PluginClassManager:
260		"""
261		Manages the plugin classes, which can be used to initialise and activate new plugins.
262
263		Loads all plugin classes from entry_point "groundwork.plugin" automatically during own initialisation.
264		Provides functions to register new plugin classes during runtime.
265		"""
266
267		def __init__(self, app, strict=False):
268		self._log = logging.getLogger(__name__)
269		self._app = app
270		self._strict = strict
271		self._classes = {}
272		self._get_plugins_by_entry_points()
273
274		def _get_plugins_by_entry_points(self):
275		"""
276		Registers plugin classes, which are in sys.path and have an entry_point called 'groundwork.plugin'.
277		:return: dict of plugin classes
278		"""
279		# Let's find and register every plugin, which is in sys.path and has defined a entry_point 'groundwork.plugin'
280		# in it's setup.py
281		entry_points = []
282		classes = {}
283		for entry_point in iter_entry_points(group='groundwork.plugin', name=None):
284		entry_points.append(entry_point)
285
286		for entry_point in entry_points:
287		try:
288		entry_point_object = entry_point.load()
289		except Exception as e:
290		# We should not throw an exception now, because a package/entry_point can be outdated, using an old
291		# api from groundwork, tries to import unavailable packages, what ever...
292		# We just do not make it available. That's all we can do.
293		self._log.warning("Couldn't load entry_point %s. Reason: %s" % (entry_point.name, e))
294		continue
295
296		if not issubclass(entry_point_object, GwBasePattern):
297		self._log.warning("entry_point %s is not a subclass of groundworkPlugin" % entry_point.name)
298		continue
299		plugin_name = entry_point_object.__name__
300
301		plugin_class = self.register_class(entry_point_object, plugin_name,
302		entrypoint_name=entry_point.name,
303		distribution_path=entry_point.dist.location,
304		distribution_key=entry_point.dist.key,
305		distribution_version=entry_point.dist.version)
306
307		classes[plugin_name] = plugin_class
308		# classes[plugin_name] = {
309		# "name": plugin_name,
310		# "entry_point": entry_point.name,
311		# "path": entry_point.dist.location,
312		# "class": entry_point_object,
313		# "distribution": {
314		# "key": entry_point.dist.key,
315		# "version": entry_point.dist.version
316		# },
317		# }
318		self._log.debug("Found plugin: %s at entry_point %s of package %s (%s)" % (plugin_name, entry_point.name,
319		entry_point.dist.key,
320		entry_point.dist.version))
321		return classes
322
323		def register(self, classes=[]):
324		"""
325		Registers new plugins.
326
327		The registration only creates a new entry for a plugin inside the _classes dictionary.
328		It does not activate or even initialise the plugin.
329
330		A plugin must be a class, which inherits directly or indirectly from GwBasePattern.
331
332		:param classes: List of plugin classes
333		:type classes: list
334		"""
335		if not isinstance(classes, list):
336		raise AttributeError("plugins must be a list, not %s." % type(classes))
337
338		plugin_registered = []
339
340		for plugin_class in classes:
341		plugin_name = plugin_class.__name__
342		self.register_class(plugin_class, plugin_name)
343		self._log.debug("Plugin %s registered" % plugin_name)
344		plugin_registered.append(plugin_name)
345
346		self._log.info("Plugins registered: %s" % ", ".join(plugin_registered))
347
348		def register_class(self, clazz, name=None, entrypoint_name=None, distribution_path=None,
349		distribution_key=None, distribution_version=None):
350
351		if name is None:
352		name = clazz.__name__
353
354		if not inspect.isclass(clazz) or not issubclass(clazz, GwBasePattern):
355		self._log.error("Given plugin is not a subclass of groundworkPlugin.")
356		if self._strict:
357		raise AttributeError("Given plugin is not a subclass of groundworkPlugin.")
358
359		if isinstance(clazz, GwBasePattern):
360		self._log.error("Given plugin %s is already initialised. Please provide a class not an instance.")
361		if self._strict:
362		raise Exception("Given plugin %s is already initialised. Please provide a class not an instance.")
363
364		if name in self._classes.keys():
365		self._log.warning("Plugin %s already registered" % name)
366		if self._strict:
367		raise PluginRegistrationException("Plugin %s already registered" % name)
368
369		self._classes[name] = PluginClass(name, clazz, entrypoint_name, distribution_path,
370		distribution_key, distribution_version)
371
372		return self._classes[name]
373
374		# self._classes[name] = {
375		# "name": plugin_name,
376		# "entry_point": None,
377		# "path": None,
378		# "class": plugin,
379		# "distribution": None,
380		# }
381
382		def get(self, name=None):
383		"""
384		Returns the plugin class object with the given name.
385		Or if a name is not given, the complete plugin dictionary is returned.
386
387		:param name: Name of a plugin
388		:return: None, single plugin or dictionary of plugins
389		"""
390		if name is None:
391		return self._classes
392		else:
393		if name not in self._classes.keys():
394		return None
395		else:
396		return self._classes[name]
397
398		def exist(self, name):
399		"""
400		Returns True if plugin class exists.
401		:param name: plugin name
402		:return: boolean
403		"""
404		if name in self._classes.keys():
405		return True
406		return False
407
408
409		class PluginClass:
410		def __init__(self, name, clazz, entrypoint_name, distribution_path, distribution_key, distribution_version):
411		self.name = name
412		self.entrypoint_name = entrypoint_name
413		self.clazz = clazz
414		self.distribution = {
415		"path": distribution_path,
416		"key": distribution_key,
417		"version": distribution_version
418		}
419

useblocks / groundwork

Push — master ( 754800...aae51d )

PluginManager.initialise() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like