WidgetPluginManager.create_widget() - Code Metrics - Inspection of "refactor widgets, fix tests" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1-unstable ( f650fd...fbeeec )

by Felipe A.

created 2016-11-11 14:48 UTC

WidgetPluginManager.create_widget() C

↳ Parent: WidgetPluginManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
dl	0
loc	25
rs	5.5
c	0
b	0
f	0

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

import sys
import argparse
import warnings
import collections

from . import mimetype


class PluginNotFoundError(ImportError):
    pass


class WidgetException(Exception):
    pass


class WidgetParameterException(WidgetException):
    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 WidgetPluginManager(RegistrablePluginManager):
    widget_types = {
        'base': collections.namedtuple(
            'Widget',
            ('place', 'type')),
        'link': collections.namedtuple(
            'Link',
            ('place', 'type', 'css', 'icon', 'text', 'endpoint', 'href')),
        'button': collections.namedtuple(
            'Button',
            ('place', 'type', 'css', 'text', 'endpoint', 'href')),
        'upload': collections.namedtuple(
            'Upload',
            ('place', 'type', 'css', 'text', 'endpoint', 'action')),
        'stylesheet': collections.namedtuple(
            'Stylesheet',
            ('place', 'type', 'endpoint', 'filename', 'href')),
        'script': collections.namedtuple(
            'Script',
            ('place', 'type', 'endpoint', 'filename', 'src')),
        'html': collections.namedtuple(
            'Html',
            ('place', 'type', 'html')),
    }

    def clear(self):
        self._widgets = []
        super(WidgetPluginManager, self).clear()

    def get_widgets(self, file=None, place=None):
        return list(self.iter_widgets(file, place))

    def iter_widgets(self, file=None, place=None):
        for filter, cwidget in self._widgets:
            try:
                if filter and not filter(file):
                    continue
            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
                    )
                continue
            if place and place != cwidget.place:
                continue
            yield cwidget

    def create_widget(self, place=None, type=None, **kwargs):
        widget_class = self.widget_types.get(type, self.widget_types['base'])
        kwargs.update(
            place=place,
            type=type
            )
        kwargs.update(
            (f, None)
            for f in widget_class._fields
            if f not in kwargs
            )
        try:
            element = widget_class(**kwargs)
        except TypeError as e:
            message = e.args[0] if e.args else ''
            if (
              'unexpected keyword argument' in message or
              'required positional argument' in message
              ):
                raise WidgetParameterException(
                    'type %s; %s; available: %r'
                    % (type, message, widget_class._fields)
                    )
            raise e
        return element

    def register_widget(self, place=None, type=None, widget=None, filter=None,
                        **kwargs):
        element = widget or self.create_widget(place, type, **kwargs)
        self._widgets.append((filter, element))


class MimetypePluginManager(RegistrablePluginManager):
    _default_mimetype_functions = (
        mimetype.by_python,
        mimetype.by_file,
        mimetype.by_default,
    )

    def clear(self):
        self._mimetype_functions = list(self._default_mimetype_functions)
        super(MimetypePluginManager, 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 register_mimetype_function(self, fnc):
        self._mimetype_functions.insert(0, fnc)


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

    def load_arguments(self, argv, base=None):

        plugin_parser = argparse.ArgumentParser(add_help=False)
        plugin_parser.add_argument(
            '--plugin',
            type=lambda x: x.split(',') if x else [],
            default=[]
            )
        parser = argparse.ArgumentParser(
            parents=(base or plugin_parser,),
            add_help=False
            )
        for plugin in plugin_parser.parse_known_args(argv)[0].plugin:
            module = self.import_plugin(plugin)
            if hasattr(module, 'register_arguments'):
                manager = ArgumentPluginManager()
                module.register_arguments(manager)
                group = parser.add_argument_group('%s arguments' % plugin)
                for argargs, argkwargs in manager._argparse_argkwargs:
                    group.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, WidgetPluginManager,
                    MimetypePluginManager, ArgumentPluginManager):
    pass


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import sys
5			import argparse
6			import warnings
7			import collections
8
9			from . import mimetype
10
11
12			class PluginNotFoundError(ImportError):
13			pass
14
15
16			class WidgetException(Exception):
17			pass
18
19
20			class WidgetParameterException(WidgetException):
21			pass
22
23
24			class PluginManagerBase(object):
25
26			@property
27			def namespaces(self):
28			return self.app.config['plugin_namespaces'] if self.app else []
29
30			def __init__(self, app=None):
31			if app is None:
32			self.clear()
33			else:
34			self.init_app(app)
35
36			def init_app(self, app):
37			self.app = app
38			if not hasattr(app, 'extensions'):
39			app.extensions = {}
40			app.extensions['plugin_manager'] = self
41			self.reload()
42
43			def reload(self):
44			self.clear()
45			for plugin in self.app.config.get('plugin_modules', ()):
46			self.load_plugin(plugin)
47
48			def clear(self):
49			pass
50
51			def import_plugin(self, plugin):
52			names = [
53			'%s.%s' % (namespace, plugin) if namespace else plugin
54			for namespace in self.namespaces
55			]
56
57			for name in names:
58			if name in sys.modules:
59			return sys.modules[name]
60
61			for name in names:
62			try:
63			__import__(name)
64			return sys.modules[name]
65			except (ImportError, KeyError):
66			pass
67
68			raise PluginNotFoundError(
69			'No plugin module %r found, tried %r' % (plugin, names),
70			plugin, names)
71
72			def load_plugin(self, plugin):
73			return self.import_plugin(plugin)
74
75
76			class RegistrablePluginManager(PluginManagerBase):
77			def load_plugin(self, plugin):
78			module = super(RegistrablePluginManager, self).load_plugin(plugin)
79			if hasattr(module, 'register_plugin'):
80			module.register_plugin(self)
81			return module
82
83
84			class BlueprintPluginManager(RegistrablePluginManager):
85			'''
86			Note: blueprints are not removed on `clear` nor reloaded on `reload`
87			as flask does not allow it.
88			'''
89			def __init__(self, app=None):
90			self._blueprint_known = set()
91			super(BlueprintPluginManager, self).__init__(app=app)
92
93			def register_blueprint(self, blueprint):
94			if blueprint not in self._blueprint_known:
95			self.app.register_blueprint(blueprint)
96			self._blueprint_known.add(blueprint)
97
98
99			class WidgetPluginManager(RegistrablePluginManager):
100			widget_types = {
101			'base': collections.namedtuple(
102			'Widget',
103			('place', 'type')),
104			'link': collections.namedtuple(
105			'Link',
106			('place', 'type', 'css', 'icon', 'text', 'endpoint', 'href')),
107			'button': collections.namedtuple(
108			'Button',
109			('place', 'type', 'css', 'text', 'endpoint', 'href')),
110			'upload': collections.namedtuple(
111			'Upload',
112			('place', 'type', 'css', 'text', 'endpoint', 'action')),
113			'stylesheet': collections.namedtuple(
114			'Stylesheet',
115			('place', 'type', 'endpoint', 'filename', 'href')),
116			'script': collections.namedtuple(
117			'Script',
118			('place', 'type', 'endpoint', 'filename', 'src')),
119			'html': collections.namedtuple(
120			'Html',
121			('place', 'type', 'html')),
122			}
123
124			def clear(self):
125			self._widgets = []
126			super(WidgetPluginManager, self).clear()
127
128			def get_widgets(self, file=None, place=None):
129			return list(self.iter_widgets(file, place))
130
131			def iter_widgets(self, file=None, place=None):
132			for filter, cwidget in self._widgets:
133			try:
134			if filter and not filter(file):
135			continue
136			except BaseException as e:
137			# Exception is handled as this method execution is deffered,
138			# making hard to debug for plugin developers.
139			warnings.warn(
140			'Plugin action filtering failed with error: %s' % e,
141			RuntimeWarning
142			)
143			continue
144			if place and place != cwidget.place:
145			continue
146			yield cwidget
147
148			def create_widget(self, place=None, type=None, **kwargs):
149			widget_class = self.widget_types.get(type, self.widget_types['base'])
150			kwargs.update(
151			place=place,
152			type=type
153			)
154			kwargs.update(
155			(f, None)
156			for f in widget_class._fields
157			if f not in kwargs
158			)
159			try:
160			element = widget_class(**kwargs)
161			except TypeError as e:
162			message = e.args[0] if e.args else ''
163			if (
164			'unexpected keyword argument' in message or
165			'required positional argument' in message
166			):
167			raise WidgetParameterException(
168			'type %s; %s; available: %r'
169			% (type, message, widget_class._fields)
170			)
171			raise e
172			return element
173
174			def register_widget(self, place=None, type=None, widget=None, filter=None,
175			**kwargs):
176			element = widget or self.create_widget(place, type, **kwargs)
177			self._widgets.append((filter, element))
178
179
180			class MimetypePluginManager(RegistrablePluginManager):
181			_default_mimetype_functions = (
182			mimetype.by_python,
183			mimetype.by_file,
184			mimetype.by_default,
185			)
186
187			def clear(self):
188			self._mimetype_functions = list(self._default_mimetype_functions)
189			super(MimetypePluginManager, self).clear()
190
191			def get_mimetype(self, path):
192			for fnc in self._mimetype_functions:
193			mime = fnc(path)
194			if mime:
195			return mime
196			return mimetype.by_default(path)
197
198			def register_mimetype_function(self, fnc):
199			self._mimetype_functions.insert(0, fnc)
200
201
202			class ArgumentPluginManager(PluginManagerBase):
203			_argparse_kwargs = {'add_help': False}
204			_argparse_arguments = argparse.Namespace()
205
206			def load_arguments(self, argv, base=None):
207
208			plugin_parser = argparse.ArgumentParser(add_help=False)
209			plugin_parser.add_argument(
210			'--plugin',
211			type=lambda x: x.split(',') if x else [],
212			default=[]
213			)
214			parser = argparse.ArgumentParser(
215			parents=(base or plugin_parser,),
216			add_help=False
217			)
218			for plugin in plugin_parser.parse_known_args(argv)[0].plugin:
219			module = self.import_plugin(plugin)
220			if hasattr(module, 'register_arguments'):
221			manager = ArgumentPluginManager()
222			module.register_arguments(manager)
223			group = parser.add_argument_group('%s arguments' % plugin)
224			for argargs, argkwargs in manager._argparse_argkwargs:
225			group.add_argument(argargs, *argkwargs)
226			self._argparse_arguments = parser.parse_args(argv)
227			return self._argparse_arguments
228
229			def clear(self):
230			self._argparse_argkwargs = []
231			super(ArgumentPluginManager, self).clear()
232
233			def register_argument(self, args, *kwargs):
234			self._argparse_argkwargs.append((args, kwargs))
235
236			def get_argument(self, name, default=None):
237			return getattr(self._argparse_arguments, name, default)
238
239
240			class PluginManager(BlueprintPluginManager, WidgetPluginManager,
241			MimetypePluginManager, ArgumentPluginManager):
242			pass
243

ergoithz / browsepy

Push — dev-4.1-unstable ( f650fd...fbeeec )

WidgetPluginManager.create_widget() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like