PluginManagerBase.load_plugin() - Code Metrics - Inspection of "add plugin arguments" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1 ( dbd54d...c25dff )

by Felipe A.

created 2016-10-25 14:25 UTC

PluginManagerBase.load_plugin() A

↳ Parent: PluginManagerBase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	1
dl	0
loc	2
rs	10
c	1
b	1
f	0

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
import collections
import argparse
import warnings

from . import mimetype
from . import widget
from .compat import isnonstriterable


class PluginNotFoundError(ImportError):
    pass


class PluginManagerBase(object):

    @property
    def namespaces(self):
        return self.app.config['plugin_namespaces'] if self.app else []

    def __init__(self, app=None):
        if app is None:
            self.clear()
        else:
            self.init_app(app)

    def init_app(self, app):
        self.app = app
        if not hasattr(app, 'extensions'):
            app.extensions = {}
        app.extensions['plugin_manager'] = self
        self.reload()

    def reload(self):
        self.clear()
        for plugin in self.app.config.get('plugin_modules', ()):
            self.load_plugin(plugin)

    def clear(self):
        pass

    def import_plugin(self, plugin):
        names = [
            '%s.%s' % (namespace, plugin) if namespace else plugin
            for namespace in self.namespaces
            ]

        for name in names:
            if name in sys.modules:
                return sys.modules[name]

        for name in names:
            try:
                __import__(name)
                return sys.modules[name]
            except (ImportError, KeyError):
                pass

        raise PluginNotFoundError(
            'No plugin module %r found, tried %r' % (plugin, names),
            plugin, names)

    def load_plugin(self, plugin):
        return self.import_plugin(plugin)


class RegistrablePluginManager(PluginManagerBase):
    def load_plugin(self, plugin):
        module = super(RegistrablePluginManager, self).load_plugin(plugin)
        if hasattr(module, 'register_plugin'):
            module.register_plugin(self)
        return module


class BlueprintPluginManager(RegistrablePluginManager):
    '''
    Note: blueprints are not removed on `clear`nor reloaded on `reload` 
    as flask does not allow it.
    '''
    def __init__(self, app=None):
        self._blueprint_known = set()
        super(BlueprintPluginManager, self).__init__(app=app)

    def register_blueprint(self, blueprint):
        if blueprint not in self._blueprint_known:
            self.app.register_blueprint(blueprint)
            self._blueprint_known.add(blueprint)


class ActionPluginManager(RegistrablePluginManager):
    action_class = collections.namedtuple(
        'CallbackAction', ('endpoint', 'widget'))
    button_class = widget.ButtonWidget
    style_class = widget.StyleWidget
    javascript_class = widget.JavascriptWidget
    link_class = widget.LinkWidget

    def clear(self):
        self._action_widgets = {}
        self._action_callback = []
        super(ActionPluginManager, self).clear()

    def get_actions(self, file):
        return list(self.iter_actions(file))

    def iter_actions(self, file):
        for callback, endpoint, cwidget in self._action_callback:
            try:
                check = callback(file)
            except BaseException as e:
                # Exception is handled  as this method execution is deffered,
                # making hard to debug for plugin developers.
                warnings.warn(
                    'Plugin action filtering failed with error: %s' % e,
                    RuntimeWarning
                    )
            else:
                if check:
                    yield self.action_class(endpoint, cwidget.for_file(file))

    def get_widgets(self, place):
        return self._action_widgets.get(place, [])

    def register_widget(self, widget):
        self._action_widgets.setdefault(widget.place, []).append(widget)

    def register_action(self, endpoint, widget, callback=None, **kwargs):
        if callable(callback):
            self._action_callback.append((callback, endpoint, widget))


class MimetypeActionPluginManager(ActionPluginManager):
    action_class = collections.namedtuple(
        'MimetypeAction', ('endpoint', 'widget'))

    _default_mimetype_functions = (
        mimetype.by_python,
        mimetype.by_file,
        mimetype.by_default,
    )

    def clear(self):
        self._mimetype_root = {}  # mimetype tree root node
        self._mimetype_functions = list(self._default_mimetype_functions)
        super(MimetypeActionPluginManager, self).clear()

    def get_mimetype(self, path):
        for fnc in self._mimetype_functions:
            mime = fnc(path)
            if mime:
                return mime
        return mimetype.by_default(path)

    def iter_actions(self, file):
        for action in super(MimetypeActionPluginManager, self)\
                        .iter_actions(file):
            yield action

        category, variant = file.mimetype.split('/')
        for tree_category in (category, '*'):
            for tree_variant in (variant, '*'):
                acts = self._mimetype_root\
                    .get(tree_category, {})\
                    .get(tree_variant, ())
                for endpoint, cwidget in acts:
                    yield self.action_class(endpoint, cwidget.for_file(file))

    def register_mimetype_function(self, fnc):
        self._mimetype_functions.insert(0, fnc)

    def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
        if not mimetypes:
            super(MimetypeActionPluginManager, self)\
                    .register_action(endpoint, widget, **kwargs)
            return
        mimetypes = mimetypes if isnonstriterable(mimetypes) else (mimetypes,)
        action = (endpoint, widget)
        for mime in mimetypes:
            category, variant = mime.split('/')
            self._mimetype_root.setdefault(
                category, {}
                ).setdefault(variant, []).append(action)


class ArgumentPluginManager(PluginManagerBase):
    _argparse_kwargs = {'add_help': False}
    _argparse_arguments = argparse.Namespace()

    def load_arguments(self, argv, base):
        parser = argparse.ArgumentParser(
            parents=(base,),
            add_help=False
            )
        for plugin in base.parse_known_args(argv)[0].plugin:
            module = self.import_plugin(plugin)
            if hasattr(module, 'register_arguments'):
                module.register_arguments(self)
        for argargs, argkwargs in self._argparse_argkwargs:
            parser.add_argument(*argargs, **argkwargs)
        self._argparse_arguments = parser.parse_args(argv)
        return self._argparse_arguments

    def clear(self):
        self._argparse_argkwargs = []
        super(ArgumentPluginManager, self).clear()

    def register_argument(self, *args, **kwargs):
        self._argparse_argkwargs.append((args, kwargs))

    def get_argument(self, name, default=None):
        return getattr(self._argparse_arguments, name, default)


class PluginManager(BlueprintPluginManager, MimetypeActionPluginManager,
                    ArgumentPluginManager):
    pass


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import sys
5			import collections
6			import argparse
7			import warnings
8
9			from . import mimetype
10			from . import widget
11			from .compat import isnonstriterable
12
13
14			class PluginNotFoundError(ImportError):
15			pass
16
17
18			class PluginManagerBase(object):
19
20			@property
21			def namespaces(self):
22			return self.app.config['plugin_namespaces'] if self.app else []
23
24			def __init__(self, app=None):
25			if app is None:
26			self.clear()
27			else:
28			self.init_app(app)
29
30			def init_app(self, app):
31			self.app = app
32			if not hasattr(app, 'extensions'):
33			app.extensions = {}
34			app.extensions['plugin_manager'] = self
35			self.reload()
36
37			def reload(self):
38			self.clear()
39			for plugin in self.app.config.get('plugin_modules', ()):
40			self.load_plugin(plugin)
41
42			def clear(self):
43			pass
44
45			def import_plugin(self, plugin):
46			names = [
47			'%s.%s' % (namespace, plugin) if namespace else plugin
48			for namespace in self.namespaces
49			]
50
51			for name in names:
52			if name in sys.modules:
53			return sys.modules[name]
54
55			for name in names:
56			try:
57			__import__(name)
58			return sys.modules[name]
59			except (ImportError, KeyError):
60			pass
61
62			raise PluginNotFoundError(
63			'No plugin module %r found, tried %r' % (plugin, names),
64			plugin, names)
65
66			def load_plugin(self, plugin):
67			return self.import_plugin(plugin)
68
69
70			class RegistrablePluginManager(PluginManagerBase):
71			def load_plugin(self, plugin):
72			module = super(RegistrablePluginManager, self).load_plugin(plugin)
73			if hasattr(module, 'register_plugin'):
74			module.register_plugin(self)
75			return module
76
77
78			class BlueprintPluginManager(RegistrablePluginManager):
79			'''
80			Note: blueprints are not removed on `clear`nor reloaded on `reload`
81			as flask does not allow it.
82			'''
83			def __init__(self, app=None):
84			self._blueprint_known = set()
85			super(BlueprintPluginManager, self).__init__(app=app)
86
87			def register_blueprint(self, blueprint):
88			if blueprint not in self._blueprint_known:
89			self.app.register_blueprint(blueprint)
90			self._blueprint_known.add(blueprint)
91
92
93			class ActionPluginManager(RegistrablePluginManager):
94			action_class = collections.namedtuple(
95			'CallbackAction', ('endpoint', 'widget'))
96			button_class = widget.ButtonWidget
97			style_class = widget.StyleWidget
98			javascript_class = widget.JavascriptWidget
99			link_class = widget.LinkWidget
100
101			def clear(self):
102			self._action_widgets = {}
103			self._action_callback = []
104			super(ActionPluginManager, self).clear()
105
106			def get_actions(self, file):
107			return list(self.iter_actions(file))
108
109			def iter_actions(self, file):
110			for callback, endpoint, cwidget in self._action_callback:
111			try:
112			check = callback(file)
113			except BaseException as e:
114			# Exception is handled as this method execution is deffered,
115			# making hard to debug for plugin developers.
116			warnings.warn(
117			'Plugin action filtering failed with error: %s' % e,
118			RuntimeWarning
119			)
120			else:
121			if check:
122			yield self.action_class(endpoint, cwidget.for_file(file))
123
124			def get_widgets(self, place):
125			return self._action_widgets.get(place, [])
126
127			def register_widget(self, widget):
128			self._action_widgets.setdefault(widget.place, []).append(widget)
129
130			def register_action(self, endpoint, widget, callback=None, **kwargs):
131			if callable(callback):
132			self._action_callback.append((callback, endpoint, widget))
133
134
135			class MimetypeActionPluginManager(ActionPluginManager):
136			action_class = collections.namedtuple(
137			'MimetypeAction', ('endpoint', 'widget'))
138
139			_default_mimetype_functions = (
140			mimetype.by_python,
141			mimetype.by_file,
142			mimetype.by_default,
143			)
144
145			def clear(self):
146			self._mimetype_root = {} # mimetype tree root node
147			self._mimetype_functions = list(self._default_mimetype_functions)
148			super(MimetypeActionPluginManager, self).clear()
149
150			def get_mimetype(self, path):
151			for fnc in self._mimetype_functions:
152			mime = fnc(path)
153			if mime:
154			return mime
155			return mimetype.by_default(path)
156
157			def iter_actions(self, file):
158			for action in super(MimetypeActionPluginManager, self)\
159			.iter_actions(file):
160			yield action
161
162			category, variant = file.mimetype.split('/')
163			for tree_category in (category, '*'):
164			for tree_variant in (variant, '*'):
165			acts = self._mimetype_root\
166			.get(tree_category, {})\
167			.get(tree_variant, ())
168			for endpoint, cwidget in acts:
169			yield self.action_class(endpoint, cwidget.for_file(file))
170
171			def register_mimetype_function(self, fnc):
172			self._mimetype_functions.insert(0, fnc)
173
174			def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
175			if not mimetypes:
176			super(MimetypeActionPluginManager, self)\
177			.register_action(endpoint, widget, **kwargs)
178			return
179			mimetypes = mimetypes if isnonstriterable(mimetypes) else (mimetypes,)
180			action = (endpoint, widget)
181			for mime in mimetypes:
182			category, variant = mime.split('/')
183			self._mimetype_root.setdefault(
184			category, {}
185			).setdefault(variant, []).append(action)
186
187
188			class ArgumentPluginManager(PluginManagerBase):
189			_argparse_kwargs = {'add_help': False}
190			_argparse_arguments = argparse.Namespace()
191
192			def load_arguments(self, argv, base):
193			parser = argparse.ArgumentParser(
194			parents=(base,),
195			add_help=False
196			)
197			for plugin in base.parse_known_args(argv)[0].plugin:
198			module = self.import_plugin(plugin)
199			if hasattr(module, 'register_arguments'):
200			module.register_arguments(self)
201			for argargs, argkwargs in self._argparse_argkwargs:
202			parser.add_argument(argargs, *argkwargs)
203			self._argparse_arguments = parser.parse_args(argv)
204			return self._argparse_arguments
205
206			def clear(self):
207			self._argparse_argkwargs = []
208			super(ArgumentPluginManager, self).clear()
209
210			def register_argument(self, args, *kwargs):
211			self._argparse_argkwargs.append((args, kwargs))
212
213			def get_argument(self, name, default=None):
214			return getattr(self._argparse_arguments, name, default)
215
216
217			class PluginManager(BlueprintPluginManager, MimetypeActionPluginManager,
218			ArgumentPluginManager):
219			pass
220

ergoithz / browsepy

Push — dev-4.1 ( dbd54d...c25dff )

PluginManagerBase.load_plugin() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like