WidgetPluginManager.clear() - Code Metrics - Inspection of "add support for widget callable properties" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1-unstable ( fbeeec...3dbdf3 )

by Felipe A.

created 2016-11-14 10:50 UTC

WidgetPluginManager.clear() A

↳ Parent: WidgetPluginManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

import sys
import argparse
import warnings
import collections

from . import mimetype


def defaultsnamedtuple(name, fields, defaults=None):
    nt = collections.namedtuple(name, fields)
    nt.__new__.__defaults__ = (None,) * len(nt._fields)
    if isinstance(defaults, collections.Mapping):
        nt.__new__.__defaults__ = tuple(nt(**defaults))
    elif defaults:
        nt.__new__.__defaults__ = tuple(nt(*defaults))
    return nt


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': defaultsnamedtuple(
            'Widget',
            ('place', 'type')),
        'link': defaultsnamedtuple(
            'Link',
            ('place', 'type', 'css', 'icon', 'text', 'endpoint', 'href'),
            {
                'text': lambda f: f.name,
                'icon': lambda f: 'dir-icon' if f.is_directory else 'file-icon'
                }),
        'button': defaultsnamedtuple(
            'Button',
            ('place', 'type', 'css', 'text', 'endpoint', 'href')),
        'upload': defaultsnamedtuple(
            'Upload',
            ('place', 'type', 'css', 'text', 'endpoint', 'action')),
        'stylesheet': defaultsnamedtuple(
            'Stylesheet',
            ('place', 'type', 'endpoint', 'filename', 'href')),
        'script': defaultsnamedtuple(
            'Script',
            ('place', 'type', 'endpoint', 'filename', 'src')),
        'html': defaultsnamedtuple(
            '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 file and 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 self._resolve_widget(file, cwidget)

    def _resolve_widget(self, file, widget):
        if file and any(map(callable, widget)):
            return widget.__class__(*[
                value(file) if callable(value) else value
                for value in widget
                ])
        return widget

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

ergoithz / browsepy

Push — dev-4.1-unstable ( fbeeec...3dbdf3 )

WidgetPluginManager.clear() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like