PluginRegistry.get_writers() - Code Metrics - Inspection of "more prometheus related changes - refactoring for..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — prom (d3fc9e)

by Kenny

created 2017-03-01 08:01 UTC

PluginRegistry.get_writers() A

↳ Parent: PluginRegistry

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	6
rs	9.4285
cc	1

# -*- coding: utf-8 -*-
"""

Plugin loader and Registry.

Currently supports the notion of Reader, Render and Writer plugins.

.. moduleauthor:: Kenny Freeman <[email protected]>

"""
import os
import imp
import inspect
import traceback
from plumd.config import Conf
from plumd.config import find as find_configs
from plumd import ConfigError, PluginLoadError
from plumd import Reader, Writer, Render

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__version__ = '0.7'
__license__ = "ISCL"
__docformat__ = 'reStructuredText'


class PluginRegistry(object):
    """Class loader and registry to maintain references.

    Uses a dict to record classes:

        class name => ( file, module, class )

    The plugin is loaded from <file> if defined in the plugin configuration
    otherwise <module> must be defined.

    If both <file> and <module> are defined in a plugin configuration, <file> is
    used and <module> ignored.
    """

    def __init__(self, log, config):
        """Plugin/Class loader and registry for reader/writer plugins.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: A plumd.Config object
        :type config: plumd.config.Conf
        """
        self.log = log
        self.defaults = {
            'interval': config.get('interval'),
            'poll.interval': config.get('poll.interval'),
            'delay.poll': config.get('delay.poll'),
            'meta': config.get('meta'),
            'enabled': True
        }
        self.classes = {}
        self.readers = {}
        self.renders = {}
        self.writers = {}

        # ensure the configured plugin directory exists
        self.plugin_dir = config.get("config.plugins")
        if not os.path.isdir(self.plugin_dir):
            msg = "invalid plugin directory configured: {0}"
            raise ConfigError(msg.format(self.plugin_dir))

        # load all configured plugins
        self.load_all()

    @property
    def nreaders(self):
        """Return the count of Reader plugins loaded.

        :rtype: int
        """
        return len(self.readers.keys())

    @property
    def nrenders(self):
        """Return the count of Render plugins loaded.

        :rtype: int
        """
        return len(self.renders.keys())

    @property
    def nwriters(self):
        """Return the count of Writer plugins loaded.

        :rtype: int
        """
        return len(self.writers.keys())

    @property
    def nplugins(self):
        """Return the count of Reader + Writer plugins loaded.

        :rtype: int
        """
        return len(self.readers.keys()) + len(self.renders.keys()) + len(self.writers.keys())

    def get_readers(self):
        """Return a list of tuples of our reader plugins: [(name, Reader)...].

        :rtype: list((str, Reader))
        """
        return self.readers.items()

    def get_renders(self):
        """Return a list of tuples of our render plugins: [(name, Reader)...].

        :rtype: list((str, Render))
        """
        return self.renders.items()

    def get_writers(self):
        """Return a list of tuples of our writer plugins: [(name, Reader)...].

        :rtype: list((str, Writer))
        """
        return self.writers.items()

    def get_reader(self, pname):
        """Return the instance of the reader plugin identified by pname.

        :param pname: The name of the plugins class
        :type pname: str
        :raises: ValueError if pname not configured
        :rtype: object
        """
        if pname not in self.readers:
            err = "reader {0} not defined in configuration"
            raise ValueError(err.format(pname))
        return self.readers[pname]

    def get_render(self, pname):
        """Return the instance of the render plugin identified by pname.

        :param pname: The name of the plugins class
        :type pname: str
        :raises: ValueError if pname not configured
        :rtype: object
        """
        if pname not in self.renders:
            err = "renderer {0} not defined in configuration"
            raise ValueError(err.format(pname))
        return self.renders[pname]

    def get_writer(self, pname):
        """Return the instance of the writer plugin identified by pname.

        :param pname: The name of the plugins class
        :type pname: str
        :raises: ValueError if pname not configured
        :rtype: object
        """
        if pname not in self.writers:
            err = "writer {0} not defined in configuration"
            raise ValueError(err.format(pname))
        return self.writers[pname]

    def load_class(self, cname, mod):
        """Load and register class from module.

        :param cname: The class name to load
        :type cname: str
        :param mod: A module object from eg. imp.load_source or __import__
        :type mod: module
        :raises: PluginLoadError: many things can go wrong during loading
        """
        for iname, icls in inspect.getmembers(mod, inspect.isclass):
            if iname == cname:
                self.classes[cname] = icls

    def load_instance(self, cname, mod, config):
        """Load and register an instance of <class> from <module>.

        :param cname: The class name to load
        :type cname: str
        :param mod: A module object from eg. imp.load_source or __import__
        :type mod: module
        :param config: A conf configuration helper instance to pass to the class
        :type config: conf
        :raises: PluginLoadError: many things can go wrong during loading
        """
        pobj = None
        # ensure plugin has a uniq name
        pname = config.get('name', exception=True)
        if pname in self.readers or pname in self.writers:
            err = "duplicate plugin name: {0}"
            raise ConfigError(err.format(pname))

        # load the class if it's not already loaded
        if cname not in self.classes:
            self.load_class(cname, mod)

        # attempt to instantiate the plugin instance
        try:
            pobj = self.classes[cname](self.log, config)
        except Exception as exc:
            trace = traceback.format_exc()
            estr = "class {0}, module {1} raised: {2}, trace:{3}"
            eargs = [cname, mod.__name__, exc, trace]
            raise PluginLoadError(estr.format(*eargs))


        # ensure it actually instantiated an object
        if not pobj:
            err = "failed to instantiate plugin {0}: {1}"
            raise ConfigError(err.format(pname, config.path))

        # it should be one of Reader or Writer, ensure we keep a reference
        pclasses = inspect.getmro(pobj.__class__)
        if Reader in pclasses:
            self.readers[pname] = pobj
        elif Writer in pclasses:
            self.writers[pname] = pobj
        elif Render in pclasses:
            self.renders[pname] = pobj
        else:
            err = "plugin is not a Reader or Writer: {0}: {1}"
            raise ConfigError(err.format(pname, cname))

    def load_from_file(self, cname, fname, config):
        """Load an instance of a Plugin from a file.

        :param cname: The name of the plugins class
        :type cname: str
        :param fname: The full path to the source file containing cname
        :type fname: str
        :param config: The plugins configuration
        :type config: plumd.config.Conf
        """
        pname = config.get("name", exception=True)
        if not os.path.isfile(fname):
            err = "invalid source file for plugin {0}: {1} in {2}"
            raise ConfigError(err.format(pname, fname, config.path))
        mod = imp.load_source(cname, fname)
        self.load_instance(cname, mod, config)

    def load_from_module(self, cname, mname, config):
        """Load an instance of a Plugin from a file.

        :param cname: The name of the plugins class
        :type cname: str
        :param mname: The full path to the source file containing cname
        :type mname: str
        :param config: The plugins configuration
        :type config: plumd.config.Conf
        """
        # allow import of modules with dots in the name
        mod = None
        fname = None
        if "." in mname:
            fname = mname
            self.log.debug("importing {0} from {1}".format(mname, fname))
        else:
            self.log.debug("importing {0}".format(mname))

        # attempt to import the module
        try:
            mod = __import__(mname.strip(), fromlist=fname)
        except Exception as e:
            msg = "module {0} import failed: {1}".format(mname, e)
            raise PluginLoadError(msg)

        # great, now load an instance class of it
        self.load_instance(cname, mod, config)

    def load_all(self):
        """Load all plugins defined in our configuration."""
        # find all the plugin configuration files
        # recursively scan the plugin configuration directory
        for pconf_fname in find_configs(self.plugin_dir, 'yaml'):
            # create a configuration object from the plugins config file
            pconfig = Conf(pconf_fname).defaults(self.defaults)

            # first of all, every plugin config needs a uniq name
            pname = pconfig.get('name', exception=True)

            # next, plugins can be defined however disabled in configuration
            if not pconfig.get('enabled'):
                msg = "skipping disabled plugin: {0}"
                self.log.info(msg.format(pname))

            # next every plugin config must define the plugins class
            pclass = pconfig.get('class', exception=True)
            if pclass in self.classes:
                msg = "duplicate plugin class defined: {0}: {1}"
                raise ConfigError(msg.format(pconf_fname, pclass))

            # next, do we load the class from a file or module?
            fname = pconfig.get('file')
            if fname:
                self.load_from_file(pclass, fname, pconfig)
            else:
                mname = pconfig.get('module', exception=True)
                self.load_from_module(pclass, mname, pconfig)

1			# -- coding: utf-8 --
2			"""
3
4			Plugin loader and Registry.
5
6			Currently supports the notion of Reader, Render and Writer plugins.
7
8			.. moduleauthor:: Kenny Freeman <[email protected]>
9
10			"""
11			import os
12			import imp
13			import inspect
14			import traceback
15			from plumd.config import Conf
16			from plumd.config import find as find_configs
17			from plumd import ConfigError, PluginLoadError
18			from plumd import Reader, Writer, Render
19
20			__author__ = 'Kenny Freeman'
21			__email__ = '[email protected]'
22			__version__ = '0.7'
23			__license__ = "ISCL"
24			__docformat__ = 'reStructuredText'
25
26
27			class PluginRegistry(object):
28			"""Class loader and registry to maintain references.
29
30			Uses a dict to record classes:
31
32			class name => ( file, module, class )
33
34			The plugin is loaded from <file> if defined in the plugin configuration
35			otherwise <module> must be defined.
36
37			If both <file> and <module> are defined in a plugin configuration, <file> is
38			used and <module> ignored.
39			"""
40
41			def __init__(self, log, config):
42			"""Plugin/Class loader and registry for reader/writer plugins.
43
44			:param log: A logger
45			:type log: logging.RootLogger
46			:param config: A plumd.Config object
47			:type config: plumd.config.Conf
48			"""
49			self.log = log
50			self.defaults = {
51			'interval': config.get('interval'),
52			'poll.interval': config.get('poll.interval'),
53			'delay.poll': config.get('delay.poll'),
54			'meta': config.get('meta'),
55			'enabled': True
56			}
57			self.classes = {}
58			self.readers = {}
59			self.renders = {}
60			self.writers = {}
61
62			# ensure the configured plugin directory exists
63			self.plugin_dir = config.get("config.plugins")
64			if not os.path.isdir(self.plugin_dir):
65			msg = "invalid plugin directory configured: {0}"
66			raise ConfigError(msg.format(self.plugin_dir))
67
68			# load all configured plugins
69			self.load_all()
70
71			@property
72			def nreaders(self):
73			"""Return the count of Reader plugins loaded.
74
75			:rtype: int
76			"""
77			return len(self.readers.keys())
78
79			@property
80			def nrenders(self):
81			"""Return the count of Render plugins loaded.
82
83			:rtype: int
84			"""
85			return len(self.renders.keys())
86
87			@property
88			def nwriters(self):
89			"""Return the count of Writer plugins loaded.
90
91			:rtype: int
92			"""
93			return len(self.writers.keys())
94
95			@property
96			def nplugins(self):
97			"""Return the count of Reader + Writer plugins loaded.
98
99			:rtype: int
100			"""
101			return len(self.readers.keys()) + len(self.renders.keys()) + len(self.writers.keys())
102
103			def get_readers(self):
104			"""Return a list of tuples of our reader plugins: [(name, Reader)...].
105
106			:rtype: list((str, Reader))
107			"""
108			return self.readers.items()
109
110			def get_renders(self):
111			"""Return a list of tuples of our render plugins: [(name, Reader)...].
112
113			:rtype: list((str, Render))
114			"""
115			return self.renders.items()
116
117			def get_writers(self):
118			"""Return a list of tuples of our writer plugins: [(name, Reader)...].
119
120			:rtype: list((str, Writer))
121			"""
122			return self.writers.items()
123
124			def get_reader(self, pname):
125			"""Return the instance of the reader plugin identified by pname.
126
127			:param pname: The name of the plugins class
128			:type pname: str
129			:raises: ValueError if pname not configured
130			:rtype: object
131			"""
132			if pname not in self.readers:
133			err = "reader {0} not defined in configuration"
134			raise ValueError(err.format(pname))
135			return self.readers[pname]
136
137			def get_render(self, pname):
138			"""Return the instance of the render plugin identified by pname.
139
140			:param pname: The name of the plugins class
141			:type pname: str
142			:raises: ValueError if pname not configured
143			:rtype: object
144			"""
145			if pname not in self.renders:
146			err = "renderer {0} not defined in configuration"
147			raise ValueError(err.format(pname))
148			return self.renders[pname]
149
150			def get_writer(self, pname):
151			"""Return the instance of the writer plugin identified by pname.
152
153			:param pname: The name of the plugins class
154			:type pname: str
155			:raises: ValueError if pname not configured
156			:rtype: object
157			"""
158			if pname not in self.writers:
159			err = "writer {0} not defined in configuration"
160			raise ValueError(err.format(pname))
161			return self.writers[pname]
162
163			def load_class(self, cname, mod):
164			"""Load and register class from module.
165
166			:param cname: The class name to load
167			:type cname: str
168			:param mod: A module object from eg. imp.load_source or __import__
169			:type mod: module
170			:raises: PluginLoadError: many things can go wrong during loading
171			"""
172			for iname, icls in inspect.getmembers(mod, inspect.isclass):
173			if iname == cname:
174			self.classes[cname] = icls
175
176			def load_instance(self, cname, mod, config):
177			"""Load and register an instance of <class> from <module>.
178
179			:param cname: The class name to load
180			:type cname: str
181			:param mod: A module object from eg. imp.load_source or __import__
182			:type mod: module
183			:param config: A conf configuration helper instance to pass to the class
184			:type config: conf
185			:raises: PluginLoadError: many things can go wrong during loading
186			"""
187			pobj = None
188			# ensure plugin has a uniq name
189			pname = config.get('name', exception=True)
190			if pname in self.readers or pname in self.writers:
191			err = "duplicate plugin name: {0}"
192			raise ConfigError(err.format(pname))
193
194			# load the class if it's not already loaded
195			if cname not in self.classes:
196			self.load_class(cname, mod)
197
198			# attempt to instantiate the plugin instance
199			try:
200			pobj = self.classes[cname](self.log, config)
201			except Exception as exc:
202			trace = traceback.format_exc()
203			estr = "class {0}, module {1} raised: {2}, trace:{3}"
204			eargs = [cname, mod.__name__, exc, trace]
205			raise PluginLoadError(estr.format(*eargs))
			0 ignored issues – show Coding Style introduced 2016-09-10 09:25 UTC by Report Bug Copy Issue Report Usage of `` or `` arguments should usually be done with care. Generally, there is nothing wrong with usage of `` or `**` arguments. For readability of the code base, we suggest to not over-use these language constructs though. For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect. Loading history...
206
207			# ensure it actually instantiated an object
208			if not pobj:
209			err = "failed to instantiate plugin {0}: {1}"
210			raise ConfigError(err.format(pname, config.path))
211
212			# it should be one of Reader or Writer, ensure we keep a reference
213			pclasses = inspect.getmro(pobj.__class__)
214			if Reader in pclasses:
215			self.readers[pname] = pobj
216			elif Writer in pclasses:
217			self.writers[pname] = pobj
218			elif Render in pclasses:
219			self.renders[pname] = pobj
220			else:
221			err = "plugin is not a Reader or Writer: {0}: {1}"
222			raise ConfigError(err.format(pname, cname))
223
224			def load_from_file(self, cname, fname, config):
225			"""Load an instance of a Plugin from a file.
226
227			:param cname: The name of the plugins class
228			:type cname: str
229			:param fname: The full path to the source file containing cname
230			:type fname: str
231			:param config: The plugins configuration
232			:type config: plumd.config.Conf
233			"""
234			pname = config.get("name", exception=True)
235			if not os.path.isfile(fname):
236			err = "invalid source file for plugin {0}: {1} in {2}"
237			raise ConfigError(err.format(pname, fname, config.path))
238			mod = imp.load_source(cname, fname)
239			self.load_instance(cname, mod, config)
240
241			def load_from_module(self, cname, mname, config):
242			"""Load an instance of a Plugin from a file.
243
244			:param cname: The name of the plugins class
245			:type cname: str
246			:param mname: The full path to the source file containing cname
247			:type mname: str
248			:param config: The plugins configuration
249			:type config: plumd.config.Conf
250			"""
251			# allow import of modules with dots in the name
252			mod = None
253			fname = None
254			if "." in mname:
255			fname = mname
256			self.log.debug("importing {0} from {1}".format(mname, fname))
257			else:
258			self.log.debug("importing {0}".format(mname))
259
260			# attempt to import the module
261			try:
262			mod = __import__(mname.strip(), fromlist=fname)
263			except Exception as e:
264			msg = "module {0} import failed: {1}".format(mname, e)
265			raise PluginLoadError(msg)
266
267			# great, now load an instance class of it
268			self.load_instance(cname, mod, config)
269
270			def load_all(self):
271			"""Load all plugins defined in our configuration."""
272			# find all the plugin configuration files
273			# recursively scan the plugin configuration directory
274			for pconf_fname in find_configs(self.plugin_dir, 'yaml'):
275			# create a configuration object from the plugins config file
276			pconfig = Conf(pconf_fname).defaults(self.defaults)
277
278			# first of all, every plugin config needs a uniq name
279			pname = pconfig.get('name', exception=True)
280
281			# next, plugins can be defined however disabled in configuration
282			if not pconfig.get('enabled'):
283			msg = "skipping disabled plugin: {0}"
284			self.log.info(msg.format(pname))
285
286			# next every plugin config must define the plugins class
287			pclass = pconfig.get('class', exception=True)
288			if pclass in self.classes:
289			msg = "duplicate plugin class defined: {0}: {1}"
290			raise ConfigError(msg.format(pconf_fname, pclass))
291
292			# next, do we load the class from a file or module?
293			fname = pconfig.get('file')
294			if fname:
295			self.load_from_file(pclass, fname, pconfig)
296			else:
297			mname = pconfig.get('module', exception=True)
298			self.load_from_module(pclass, mname, pconfig)

s4z / plumd

Branch — prom (d3fc9e)

PluginRegistry.get_writers() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like