ArgumentPluginManager.load_arguments() - Code Metrics - Inspection of "implement head-button, multifile upload, add plugi..." - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1 ( dab2a4...5e66f2 )

by Felipe A.

created 2016-11-08 15:57 UTC

ArgumentPluginManager.load_arguments() B

↳ Parent: ArgumentPluginManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
dl	0
loc	22
rs	7.7857
c	0
b	0
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
    head_button_class = widget.HeadButtonWidget
    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=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, 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			head_button_class = widget.HeadButtonWidget
98			style_class = widget.StyleWidget
99			javascript_class = widget.JavascriptWidget
100			link_class = widget.LinkWidget
101
102			def clear(self):
103			self._action_widgets = {}
104			self._action_callback = []
105			super(ActionPluginManager, self).clear()
106
107			def get_actions(self, file):
108			return list(self.iter_actions(file))
109
110			def iter_actions(self, file):
111			for callback, endpoint, cwidget in self._action_callback:
112			try:
113			check = callback(file)
114			except BaseException as e:
115			# Exception is handled as this method execution is deffered,
116			# making hard to debug for plugin developers.
117			warnings.warn(
118			'Plugin action filtering failed with error: %s' % e,
119			RuntimeWarning
120			)
121			else:
122			if check:
123			yield self.action_class(endpoint, cwidget.for_file(file))
124
125			def get_widgets(self, place):
126			return self._action_widgets.get(place, [])
127
128			def register_widget(self, widget):
129			self._action_widgets.setdefault(widget.place, []).append(widget)
130
131			def register_action(self, endpoint, widget, callback=None, **kwargs):
132			if callable(callback):
133			self._action_callback.append((callback, endpoint, widget))
134
135
136			class MimetypeActionPluginManager(ActionPluginManager):
137			action_class = collections.namedtuple(
138			'MimetypeAction', ('endpoint', 'widget'))
139
140			_default_mimetype_functions = (
141			mimetype.by_python,
142			mimetype.by_file,
143			mimetype.by_default,
144			)
145
146			def clear(self):
147			self._mimetype_root = {} # mimetype tree root node
148			self._mimetype_functions = list(self._default_mimetype_functions)
149			super(MimetypeActionPluginManager, self).clear()
150
151			def get_mimetype(self, path):
152			for fnc in self._mimetype_functions:
153			mime = fnc(path)
154			if mime:
155			return mime
156			return mimetype.by_default(path)
157
158			def iter_actions(self, file):
159			for action in super(MimetypeActionPluginManager, self)\
160			.iter_actions(file):
161			yield action
162
163			category, variant = file.mimetype.split('/')
164			for tree_category in (category, '*'):
165			for tree_variant in (variant, '*'):
166			acts = self._mimetype_root\
167			.get(tree_category, {})\
168			.get(tree_variant, ())
169			for endpoint, cwidget in acts:
170			yield self.action_class(endpoint, cwidget.for_file(file))
171
172			def register_mimetype_function(self, fnc):
173			self._mimetype_functions.insert(0, fnc)
174
175			def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
176			if not mimetypes:
177			super(MimetypeActionPluginManager, self)\
178			.register_action(endpoint, widget, **kwargs)
179			return
180			mimetypes = mimetypes if isnonstriterable(mimetypes) else (mimetypes,)
181			action = (endpoint, widget)
182			for mime in mimetypes:
183			category, variant = mime.split('/')
184			self._mimetype_root.setdefault(
185			category, {}
186			).setdefault(variant, []).append(action)
187
188
189			class ArgumentPluginManager(PluginManagerBase):
190			_argparse_kwargs = {'add_help': False}
191			_argparse_arguments = argparse.Namespace()
192
193			def load_arguments(self, argv, base=None):
194
195			plugin_parser = argparse.ArgumentParser(add_help=False)
196			plugin_parser.add_argument(
197			'--plugin',
198			type=lambda x: x.split(',') if x else [],
199			default=[]
200			)
201			parser = argparse.ArgumentParser(
202			parents=(base or plugin_parser,),
203			add_help=False
204			)
205			for plugin in plugin_parser.parse_known_args(argv)[0].plugin:
206			module = self.import_plugin(plugin)
207			if hasattr(module, 'register_arguments'):
208			manager = ArgumentPluginManager()
209			module.register_arguments(manager)
210			group = parser.add_argument_group('%s arguments' % plugin)
211			for argargs, argkwargs in manager._argparse_argkwargs:
212			group.add_argument(argargs, *argkwargs)
213			self._argparse_arguments = parser.parse_args(argv)
214			return self._argparse_arguments
215
216			def clear(self):
217			self._argparse_argkwargs = []
218			super(ArgumentPluginManager, self).clear()
219
220			def register_argument(self, args, *kwargs):
221			self._argparse_argkwargs.append((args, kwargs))
222
223			def get_argument(self, name, default=None):
224			return getattr(self._argparse_arguments, name, default)
225
226
227			class PluginManager(BlueprintPluginManager, MimetypeActionPluginManager,
228			ArgumentPluginManager):
229			pass
230

ergoithz / browsepy

Push — dev-4.1 ( dab2a4...5e66f2 )

ArgumentPluginManager.load_arguments() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like