PluginManager.get_actions() - Code Metrics - Inspection of "first step on backwards compatibility" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1-unstable ( 90aa3b...e0128e )

by Felipe A.

created 2016-11-22 17:01 UTC

PluginManager.get_actions() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

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

import re
import sys
import argparse
import warnings
import collections

from . import mimetype
from .compat import deprecated


def defaultsnamedtuple(name, fields, defaults=None):
    '''
    Generate namedtuple with default values.

    :param name: name
    :param fields: iterable with field names
    :param defaults: iterable or mapping with field defaults
    :returns: defaultdict with given fields and given defaults
    :rtype: collections.defaultdict
    '''
    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):
    '''
    Base plugin manager for plugin module loading and Flask extension logic.
    '''

    @property
    def namespaces(self):
        '''
        List of plugin namespaces taken from app config.
        '''
        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):
        '''
        Initialize this Flask extension for given app.
        '''
        self.app = app
        if not hasattr(app, 'extensions'):
            app.extensions = {}
        app.extensions['plugin_manager'] = self
        self.reload()

    def reload(self):
        '''
        Clear plugin manager state and reload plugins.
        '''
        self.clear()
        for plugin in self.app.config.get('plugin_modules', ()):
            self.load_plugin(plugin)

    def clear(self):
        '''
        Clear plugin manager state.
        '''
        pass

    def import_plugin(self, plugin):
        '''
        Import plugin by given name, looking at :attr:`namespaces`.

        :param plugin: plugin module name
        :type plugin: str
        :raises PluginNotFoundError: if not found on any namespace
        '''
        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):
        '''
        Import plugin (see :meth:`import_plugin`) and load related data.

        :param plugin: plugin module name
        :type plugin: str
        :raises PluginNotFoundError: if not found on any namespace
        '''
        return self.import_plugin(plugin)


class RegistrablePluginManager(PluginManagerBase):
    '''
    Base plugin manager for plugin registration via :func:`register_plugin`
    functions at plugin module level.
    '''
    def load_plugin(self, plugin):
        '''
        Import plugin (see :meth:`import_plugin`) and load related data.

        If available, plugin's module-level :func:`register_plugin` function
        will be called with current plugin manager instance as first argument.

        :param plugin: plugin module name
        :type plugin: str
        :raises PluginNotFoundError: if not found on any namespace
        '''
        module = super(RegistrablePluginManager, self).load_plugin(plugin)
        if hasattr(module, 'register_plugin'):
            module.register_plugin(self)
        return module


class BlueprintPluginManager(RegistrablePluginManager):
    '''
    Manager for blueprint registration via :meth:`register_plugin` calls.

    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):
        '''
        Register given blueprint on curren app.

        This method is provided for using inside plugin's module-level
        :func:`register_plugin` functions.

        :param blueprint: blueprint object with plugin endpoints
        :type blueprint: flask.Blueprint
        '''
        if blueprint not in self._blueprint_known:
            self.app.register_blueprint(blueprint)
            self._blueprint_known.add(blueprint)


class WidgetPluginManager(RegistrablePluginManager):
    '''
    Plugin manager for widget registration.

    This class provides a dictionary of widget types at its
    :attr:`widget_types` attribute. They can be referenced by their keys on
    both :meth:`create_widget` and :meth:`register_widget` methods' `type`
    parameter, or instantiated directly and passed to :meth:`register_widget`
    via `widget` parameter.
    '''
    widget_types = {
        'base': defaultsnamedtuple(
            'Widget',
            ('place', 'type')),
        'link': defaultsnamedtuple(
            'Link',
            ('place', 'type', 'css', 'icon', 'text', 'endpoint', 'href'),
            {
                'type': 'link',
                'text': lambda f: f.name,
                'icon': lambda f: f.category
            }),
        'button': defaultsnamedtuple(
            'Button',
            ('place', 'type', 'css', 'text', 'endpoint', 'href'),
            {'type': 'button'}),
        'upload': defaultsnamedtuple(
            'Upload',
            ('place', 'type', 'css', 'text', 'endpoint', 'action'),
            {'type': 'upload'}),
        'stylesheet': defaultsnamedtuple(
            'Stylesheet',
            ('place', 'type', 'endpoint', 'filename', 'href'),
            {'type': 'stylesheet'}),
        'script': defaultsnamedtuple(
            'Script',
            ('place', 'type', 'endpoint', 'filename', 'src'),
            {'type': 'script'}),
        'html': defaultsnamedtuple(
            'Html',
            ('place', 'type', 'html'),
            {'type': 'html'}),
    }

    def clear(self):
        '''
        Clear plugin manager state.

        Registered widgets will be disposed after calling this method.
        '''
        self._widgets = []
        super(WidgetPluginManager, self).clear()

    def get_widgets(self, file=None, place=None):
        '''
        List registered widgets, optionally matching given criteria.

        :param file: optional file object will be passed to widgets' filter
                     functions.
        :type file: browsepy.file.Node or None
        :param place: optional template place hint.
        :type place: str
        :returns: list of widget instances
        :rtype: list of objects
        '''
        return list(self.iter_widgets(file, place))

    @classmethod
    def _resolve_widget(cls, file, widget):
        '''
        Resolve widget callable properties into static ones.

        :param file: file will be used to resolve callable properties.
        :type file: browsepy.file.Node
        :param widget: widget instance optionally with callable properties
        :type widget: object
        :returns: a new widget instance of the same type as widget parameter
        :rtype: object
        '''
        return widget.__class__(*[
            value(file) if callable(value) else value
            for value in widget
            ])

    def iter_widgets(self, file=None, place=None):
        '''
        Iterate registered widgets, optionally matching given criteria.

        :param file: optional file object will be passed to widgets' filter
                     functions.
        :type file: browsepy.file.Node or None
        :param place: optional template place hint.
        :type place: str
        :yields: widget instances
        :ytype: object
        '''
        for filter, dynamic, 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
            if file and dynamic:
                cwidget = self._resolve_widget(file, cwidget)
            yield cwidget

    def create_widget(self, place, type, file=None, **kwargs):
        '''
        Create a widget object based on given arguments.

        If file object is provided, callable arguments will be resolved:
        its return value will be used after calling them with file as first
        parameter.

        All extra `kwargs` parameters will be passed to widget constructor.

        :param place: place hint where widget should be shown.
        :type place: str
        :param type: widget type name as taken from :attr:`widget_types` dict
                     keys.
        :type type: str
        :param file: optional file object for widget attribute resolving
        :type type: browsepy.files.Node or None
        :returns: widget instance
        :rtype: object
        '''
        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
        if file and any(map(callable, element)):
            return self._resolve_widget(file, element)
        return element

    def register_widget(self, place=None, type=None, widget=None, filter=None,
                        **kwargs):
        '''
        Create (see :meth:`create_widget`) or use provided widget and register
        it.

        This method provides this dual behavior in order to simplify widget
        creation-registration on an functional single step without sacrifycing
        the reusability of a object-oriented approach.

        :param place: where widget should be placed. This param conflicts
                      with `widget` argument.
        :type place: str or None
        :param type: widget type name as taken from :attr:`widget_types` dict
                     keys. This param conflicts with `widget` argument.
        :type type: str or None
        :param widget: optional widget object will be used as is. This param
                       conflicts with both place and type arguments.
        :type widget: object or None
        :raises TypeError: if both widget and place or type are provided at
                           the same time (they're mutually exclusive).
        :returns: created or given widget object
        :rtype: object
        '''
        if not widget and not (place and type):
            raise TypeError(
                'register_widget takes either place and type or widget'
                )
        widget = widget or self.create_widget(place, type, **kwargs)
        dynamic = any(map(callable, widget))
        self._widgets.append((filter, dynamic, widget))
        return widget


class MimetypePluginManager(RegistrablePluginManager):
    '''
    Plugin manager for mimetype-function registration.
    '''
    _default_mimetype_functions = (
        mimetype.by_python,
        mimetype.by_file,
        mimetype.by_default,
    )

    def clear(self):
        '''
        Clear plugin manager state.

        Registered mimetype functions will be disposed after calling this
        method.
        '''
        self._mimetype_functions = list(self._default_mimetype_functions)
        super(MimetypePluginManager, self).clear()

    def get_mimetype(self, path):
        '''
        Get mimetype of given path calling all registered mime functions (and
        default ones).

        :param path: filesystem path of file
        :type path: str
        :returns: mimetype
        :rtype: str
        '''
        for fnc in self._mimetype_functions:
            mime = fnc(path)
            if mime:
                return mime
        return mimetype.by_default(path)

    def register_mimetype_function(self, fnc):
        '''
        Register mimetype function.

        Given function must accept a filesystem path as string and return
        a mimetype string or None.

        :param fnc: callable accepting a path string
        :type fnc: callable
        '''
        self._mimetype_functions.insert(0, fnc)


class ArgumentPluginManager(PluginManagerBase):
    '''
    Plugin manager for command-line argument registration.

    This function is used by browsepy's :mod:`__main__` module in order
    to attach extra arguments at argument-parsing time.

    This is done by :meth:`load_arguments` which imports all plugin modules
    and calls their respective :func:`register_arguments` module-level
    function.
    '''
    _argparse_kwargs = {'add_help': False}
    _argparse_arguments = argparse.Namespace()

    def load_arguments(self, argv, base=None):
        '''
        Process given argument list based on registered arguments and given
        optional base :class:`argparse.ArgumentParser` instance.

        This method saves processed arguments on itself, and this state won't
        be lost after :meth:`clean` calls.

        Processed argument state will be available via :meth:`get_argument`
        method.

        :param argv: command-line arguments (without command itself)
        :type argv: iterable of str
        :param base: optional base :class:`argparse.ArgumentParser` instance.
        :type base: argparse.ArgumentParser or None
        :returns: argparse.Namespace instance with processed arguments as
                  given by :meth:`argparse.ArgumentParser.parse_args`.
        :rtype: argparse.Namespace
        '''
        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):
        '''
        Clear plugin manager state.

        Registered command-line arguments will be disposed after calling this
        method.
        '''
        self._argparse_argkwargs = []
        super(ArgumentPluginManager, self).clear()

    def register_argument(self, *args, **kwargs):
        '''
        Register command-line argument.

        All given arguments will be passed directly to
        :meth:`argparse.ArgumentParser.add_argument` calls by
        :meth:`load_arguments` method.

        See :meth:`argparse.ArgumentParser.add_argument` documentation for
        further information.
        '''
        self._argparse_argkwargs.append((args, kwargs))

    def get_argument(self, name, default=None):
        '''
        Get argument value from last :meth:`load_arguments` call.

        Keep in mind :meth:`argparse.ArgumentParser.parse_args` generates
        its own command-line arguments if `dest` kwarg is not provided,
        so ie. `--my-option` became available as `my_option`.

        :param name: command-line argument name
        :type name: str
        :param default: default value if parameter is not found
        :returns: command-line argument or default value
        '''
        return getattr(self._argparse_arguments, name, default)


class PluginManager(BlueprintPluginManager, WidgetPluginManager,
                    MimetypePluginManager, ArgumentPluginManager):
    '''
    Main plugin manager

    Provides:
        * Plugin module loading and Flask extension logic.
        * Plugin registration via :func:`register_plugin` functions at plugin
          module level.
        * Plugin blueprint registration via :meth:`register_plugin` calls.
        * Widget registration via :meth:`register_widget` method.
        * Mimetype function registration via :meth:`register_mimetype_function`
          method.
        * Command-line argument registration calling :func:`register_arguments`
          at plugin module level and providing :meth:`register_argument`
          method.

    This class also provides a dictionary of widget types at its
    :attr:`widget_types` attribute. They can be referenced by their keys on
    both :meth:`create_widget` and :meth:`register_widget` methods' `type`
    parameter, or instantiated directly and passed to :meth:`register_widget`
    via `widget` parameter.
    '''
    def clear(self):
        '''
        Clear plugin manager state.

        Registered widgets will be disposed after calling this method.

        Registered mimetype functions will be disposed after calling this
        method.

        Registered command-line arguments will be disposed after calling this
        method.
        '''
        super(PluginManager, self).clear()

    @deprecated
    def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
        deprecated_places = {
            'javascript': 'scripts',
            'style': 'styles',
            'button': 'entry-actions',
            'link': 'entry-link',
            None: 'none'
            }
        widget_mimetype_re = re.compile(
            '^%s$' % '$|^'.join(
                map(re.escape, mimetypes)
                ).replace('\\*', '[^/]+')
            )
        print(widget_mimetype_re.pattern)
        widget_type = getattr(widget, '_type', 'base')
        widget_fields = self.widget_types[widget_type]._fields
        widget_props = {
            name: getattr(widget, name)
            for name in widget_fields
            if hasattr(widget, name)
            }
        widget_props.update(
            type=widget_type,
            place=deprecated_places.get(
                widget_props['place'],
                widget_props['place']
                ),
            filter=lambda f: widget_mimetype_re.match(f.type) is not None
            )
        if 'endpoint' in widget_fields:
            widget_props['endpoint'] = endpoint
        self.register_widget(**widget_props)

    @deprecated
    def get_actions(self, file):
        return self.get_widgets(file=file)


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import re
5			import sys
6			import argparse
7			import warnings
8			import collections
9
10			from . import mimetype
11			from .compat import deprecated
12
13
14			def defaultsnamedtuple(name, fields, defaults=None):
15			'''
16			Generate namedtuple with default values.
17
18			:param name: name
19			:param fields: iterable with field names
20			:param defaults: iterable or mapping with field defaults
21			:returns: defaultdict with given fields and given defaults
22			:rtype: collections.defaultdict
23			'''
24			nt = collections.namedtuple(name, fields)
25			nt.__new__.__defaults__ = (None,) * len(nt._fields)
26			if isinstance(defaults, collections.Mapping):
27			nt.__new__.__defaults__ = tuple(nt(**defaults))
28			elif defaults:
29			nt.__new__.__defaults__ = tuple(nt(*defaults))
30			return nt
31
32
33			class PluginNotFoundError(ImportError):
34			pass
35
36
37			class WidgetException(Exception):
38			pass
39
40
41			class WidgetParameterException(WidgetException):
42			pass
43
44
45			class PluginManagerBase(object):
46			'''
47			Base plugin manager for plugin module loading and Flask extension logic.
48			'''
49
50			@property
51			def namespaces(self):
52			'''
53			List of plugin namespaces taken from app config.
54			'''
55			return self.app.config['plugin_namespaces'] if self.app else []
56
57			def __init__(self, app=None):
58			if app is None:
59			self.clear()
60			else:
61			self.init_app(app)
62
63			def init_app(self, app):
64			'''
65			Initialize this Flask extension for given app.
66			'''
67			self.app = app
68			if not hasattr(app, 'extensions'):
69			app.extensions = {}
70			app.extensions['plugin_manager'] = self
71			self.reload()
72
73			def reload(self):
74			'''
75			Clear plugin manager state and reload plugins.
76			'''
77			self.clear()
78			for plugin in self.app.config.get('plugin_modules', ()):
79			self.load_plugin(plugin)
80
81			def clear(self):
82			'''
83			Clear plugin manager state.
84			'''
85			pass
86
87			def import_plugin(self, plugin):
88			'''
89			Import plugin by given name, looking at :attr:`namespaces`.
90
91			:param plugin: plugin module name
92			:type plugin: str
93			:raises PluginNotFoundError: if not found on any namespace
94			'''
95			names = [
96			'%s.%s' % (namespace, plugin) if namespace else plugin
97			for namespace in self.namespaces
98			]
99
100			for name in names:
101			if name in sys.modules:
102			return sys.modules[name]
103
104			for name in names:
105			try:
106			__import__(name)
107			return sys.modules[name]
108			except (ImportError, KeyError):
109			pass
110
111			raise PluginNotFoundError(
112			'No plugin module %r found, tried %r' % (plugin, names),
113			plugin, names)
114
115			def load_plugin(self, plugin):
116			'''
117			Import plugin (see :meth:`import_plugin`) and load related data.
118
119			:param plugin: plugin module name
120			:type plugin: str
121			:raises PluginNotFoundError: if not found on any namespace
122			'''
123			return self.import_plugin(plugin)
124
125
126			class RegistrablePluginManager(PluginManagerBase):
127			'''
128			Base plugin manager for plugin registration via :func:`register_plugin`
129			functions at plugin module level.
130			'''
131			def load_plugin(self, plugin):
132			'''
133			Import plugin (see :meth:`import_plugin`) and load related data.
134
135			If available, plugin's module-level :func:`register_plugin` function
136			will be called with current plugin manager instance as first argument.
137
138			:param plugin: plugin module name
139			:type plugin: str
140			:raises PluginNotFoundError: if not found on any namespace
141			'''
142			module = super(RegistrablePluginManager, self).load_plugin(plugin)
143			if hasattr(module, 'register_plugin'):
144			module.register_plugin(self)
145			return module
146
147
148			class BlueprintPluginManager(RegistrablePluginManager):
149			'''
150			Manager for blueprint registration via :meth:`register_plugin` calls.
151
152			Note: blueprints are not removed on `clear` nor reloaded on `reload`
153			as flask does not allow it.
154			'''
155			def __init__(self, app=None):
156			self._blueprint_known = set()
157			super(BlueprintPluginManager, self).__init__(app=app)
158
159			def register_blueprint(self, blueprint):
160			'''
161			Register given blueprint on curren app.
162
163			This method is provided for using inside plugin's module-level
164			:func:`register_plugin` functions.
165
166			:param blueprint: blueprint object with plugin endpoints
167			:type blueprint: flask.Blueprint
168			'''
169			if blueprint not in self._blueprint_known:
170			self.app.register_blueprint(blueprint)
171			self._blueprint_known.add(blueprint)
172
173
174			class WidgetPluginManager(RegistrablePluginManager):
175			'''
176			Plugin manager for widget registration.
177
178			This class provides a dictionary of widget types at its
179			:attr:`widget_types` attribute. They can be referenced by their keys on
180			both :meth:`create_widget` and :meth:`register_widget` methods' `type`
181			parameter, or instantiated directly and passed to :meth:`register_widget`
182			via `widget` parameter.
183			'''
184			widget_types = {
185			'base': defaultsnamedtuple(
186			'Widget',
187			('place', 'type')),
188			'link': defaultsnamedtuple(
189			'Link',
190			('place', 'type', 'css', 'icon', 'text', 'endpoint', 'href'),
191			{
192			'type': 'link',
193			'text': lambda f: f.name,
194			'icon': lambda f: f.category
195			}),
196			'button': defaultsnamedtuple(
197			'Button',
198			('place', 'type', 'css', 'text', 'endpoint', 'href'),
199			{'type': 'button'}),
200			'upload': defaultsnamedtuple(
201			'Upload',
202			('place', 'type', 'css', 'text', 'endpoint', 'action'),
203			{'type': 'upload'}),
204			'stylesheet': defaultsnamedtuple(
205			'Stylesheet',
206			('place', 'type', 'endpoint', 'filename', 'href'),
207			{'type': 'stylesheet'}),
208			'script': defaultsnamedtuple(
209			'Script',
210			('place', 'type', 'endpoint', 'filename', 'src'),
211			{'type': 'script'}),
212			'html': defaultsnamedtuple(
213			'Html',
214			('place', 'type', 'html'),
215			{'type': 'html'}),
216			}
217
218			def clear(self):
219			'''
220			Clear plugin manager state.
221
222			Registered widgets will be disposed after calling this method.
223			'''
224			self._widgets = []
225			super(WidgetPluginManager, self).clear()
226
227			def get_widgets(self, file=None, place=None):
228			'''
229			List registered widgets, optionally matching given criteria.
230
231			:param file: optional file object will be passed to widgets' filter
232			functions.
233			:type file: browsepy.file.Node or None
234			:param place: optional template place hint.
235			:type place: str
236			:returns: list of widget instances
237			:rtype: list of objects
238			'''
239			return list(self.iter_widgets(file, place))
240
241			@classmethod
242			def _resolve_widget(cls, file, widget):
243			'''
244			Resolve widget callable properties into static ones.
245
246			:param file: file will be used to resolve callable properties.
247			:type file: browsepy.file.Node
248			:param widget: widget instance optionally with callable properties
249			:type widget: object
250			:returns: a new widget instance of the same type as widget parameter
251			:rtype: object
252			'''
253			return widget.__class__(*[
254			value(file) if callable(value) else value
255			for value in widget
256			])
257
258			def iter_widgets(self, file=None, place=None):
259			'''
260			Iterate registered widgets, optionally matching given criteria.
261
262			:param file: optional file object will be passed to widgets' filter
263			functions.
264			:type file: browsepy.file.Node or None
265			:param place: optional template place hint.
266			:type place: str
267			:yields: widget instances
268			:ytype: object
269			'''
270			for filter, dynamic, cwidget in self._widgets:
271			try:
272			if file and filter and not filter(file):
273			continue
274			except BaseException as e:
275			# Exception is handled as this method execution is deffered,
276			# making hard to debug for plugin developers.
277			warnings.warn(
278			'Plugin action filtering failed with error: %s' % e,
279			RuntimeWarning
280			)
281			continue
282			if place and place != cwidget.place:
283			continue
284			if file and dynamic:
285			cwidget = self._resolve_widget(file, cwidget)
286			yield cwidget
287
288			def create_widget(self, place, type, file=None, **kwargs):
289			'''
290			Create a widget object based on given arguments.
291
292			If file object is provided, callable arguments will be resolved:
293			its return value will be used after calling them with file as first
294			parameter.
295
296			All extra `kwargs` parameters will be passed to widget constructor.
297
298			:param place: place hint where widget should be shown.
299			:type place: str
300			:param type: widget type name as taken from :attr:`widget_types` dict
301			keys.
302			:type type: str
303			:param file: optional file object for widget attribute resolving
304			:type type: browsepy.files.Node or None
305			:returns: widget instance
306			:rtype: object
307			'''
308			widget_class = self.widget_types.get(type, self.widget_types['base'])
309			kwargs.update(place=place, type=type)
310			try:
311			element = widget_class(**kwargs)
312			except TypeError as e:
313			message = e.args[0] if e.args else ''
314			if (
315			'unexpected keyword argument' in message or
316			'required positional argument' in message
317			):
318			raise WidgetParameterException(
319			'type %s; %s; available: %r'
320			% (type, message, widget_class._fields)
321			)
322			raise e
323			if file and any(map(callable, element)):
324			return self._resolve_widget(file, element)
325			return element
326
327			def register_widget(self, place=None, type=None, widget=None, filter=None,
328			**kwargs):
329			'''
330			Create (see :meth:`create_widget`) or use provided widget and register
331			it.
332
333			This method provides this dual behavior in order to simplify widget
334			creation-registration on an functional single step without sacrifycing
335			the reusability of a object-oriented approach.
336
337			:param place: where widget should be placed. This param conflicts
338			with `widget` argument.
339			:type place: str or None
340			:param type: widget type name as taken from :attr:`widget_types` dict
341			keys. This param conflicts with `widget` argument.
342			:type type: str or None
343			:param widget: optional widget object will be used as is. This param
344			conflicts with both place and type arguments.
345			:type widget: object or None
346			:raises TypeError: if both widget and place or type are provided at
347			the same time (they're mutually exclusive).
348			:returns: created or given widget object
349			:rtype: object
350			'''
351			if not widget and not (place and type):
352			raise TypeError(
353			'register_widget takes either place and type or widget'
354			)
355			widget = widget or self.create_widget(place, type, **kwargs)
356			dynamic = any(map(callable, widget))
357			self._widgets.append((filter, dynamic, widget))
358			return widget
359
360
361			class MimetypePluginManager(RegistrablePluginManager):
362			'''
363			Plugin manager for mimetype-function registration.
364			'''
365			_default_mimetype_functions = (
366			mimetype.by_python,
367			mimetype.by_file,
368			mimetype.by_default,
369			)
370
371			def clear(self):
372			'''
373			Clear plugin manager state.
374
375			Registered mimetype functions will be disposed after calling this
376			method.
377			'''
378			self._mimetype_functions = list(self._default_mimetype_functions)
379			super(MimetypePluginManager, self).clear()
380
381			def get_mimetype(self, path):
382			'''
383			Get mimetype of given path calling all registered mime functions (and
384			default ones).
385
386			:param path: filesystem path of file
387			:type path: str
388			:returns: mimetype
389			:rtype: str
390			'''
391			for fnc in self._mimetype_functions:
392			mime = fnc(path)
393			if mime:
394			return mime
395			return mimetype.by_default(path)
396
397			def register_mimetype_function(self, fnc):
398			'''
399			Register mimetype function.
400
401			Given function must accept a filesystem path as string and return
402			a mimetype string or None.
403
404			:param fnc: callable accepting a path string
405			:type fnc: callable
406			'''
407			self._mimetype_functions.insert(0, fnc)
408
409
410			class ArgumentPluginManager(PluginManagerBase):
411			'''
412			Plugin manager for command-line argument registration.
413
414			This function is used by browsepy's :mod:`__main__` module in order
415			to attach extra arguments at argument-parsing time.
416
417			This is done by :meth:`load_arguments` which imports all plugin modules
418			and calls their respective :func:`register_arguments` module-level
419			function.
420			'''
421			_argparse_kwargs = {'add_help': False}
422			_argparse_arguments = argparse.Namespace()
423
424			def load_arguments(self, argv, base=None):
425			'''
426			Process given argument list based on registered arguments and given
427			optional base :class:`argparse.ArgumentParser` instance.
428
429			This method saves processed arguments on itself, and this state won't
430			be lost after :meth:`clean` calls.
431
432			Processed argument state will be available via :meth:`get_argument`
433			method.
434
435			:param argv: command-line arguments (without command itself)
436			:type argv: iterable of str
437			:param base: optional base :class:`argparse.ArgumentParser` instance.
438			:type base: argparse.ArgumentParser or None
439			:returns: argparse.Namespace instance with processed arguments as
440			given by :meth:`argparse.ArgumentParser.parse_args`.
441			:rtype: argparse.Namespace
442			'''
443			plugin_parser = argparse.ArgumentParser(add_help=False)
444			plugin_parser.add_argument(
445			'--plugin',
446			type=lambda x: x.split(',') if x else [],
447			default=[]
448			)
449			parser = argparse.ArgumentParser(
450			parents=(base or plugin_parser,),
451			add_help=False
452			)
453			for plugin in plugin_parser.parse_known_args(argv)[0].plugin:
454			module = self.import_plugin(plugin)
455			if hasattr(module, 'register_arguments'):
456			manager = ArgumentPluginManager()
457			module.register_arguments(manager)
458			group = parser.add_argument_group('%s arguments' % plugin)
459			for argargs, argkwargs in manager._argparse_argkwargs:
460			group.add_argument(argargs, *argkwargs)
461			self._argparse_arguments = parser.parse_args(argv)
462			return self._argparse_arguments
463
464			def clear(self):
465			'''
466			Clear plugin manager state.
467
468			Registered command-line arguments will be disposed after calling this
469			method.
470			'''
471			self._argparse_argkwargs = []
472			super(ArgumentPluginManager, self).clear()
473
474			def register_argument(self, args, *kwargs):
475			'''
476			Register command-line argument.
477
478			All given arguments will be passed directly to
479			:meth:`argparse.ArgumentParser.add_argument` calls by
480			:meth:`load_arguments` method.
481
482			See :meth:`argparse.ArgumentParser.add_argument` documentation for
483			further information.
484			'''
485			self._argparse_argkwargs.append((args, kwargs))
486
487			def get_argument(self, name, default=None):
488			'''
489			Get argument value from last :meth:`load_arguments` call.
490
491			Keep in mind :meth:`argparse.ArgumentParser.parse_args` generates
492			its own command-line arguments if `dest` kwarg is not provided,
493			so ie. `--my-option` became available as `my_option`.
494
495			:param name: command-line argument name
496			:type name: str
497			:param default: default value if parameter is not found
498			:returns: command-line argument or default value
499			'''
500			return getattr(self._argparse_arguments, name, default)
501
502
503			class PluginManager(BlueprintPluginManager, WidgetPluginManager,
504			MimetypePluginManager, ArgumentPluginManager):
505			'''
506			Main plugin manager
507
508			Provides:
509			* Plugin module loading and Flask extension logic.
510			* Plugin registration via :func:`register_plugin` functions at plugin
511			module level.
512			* Plugin blueprint registration via :meth:`register_plugin` calls.
513			* Widget registration via :meth:`register_widget` method.
514			* Mimetype function registration via :meth:`register_mimetype_function`
515			method.
516			* Command-line argument registration calling :func:`register_arguments`
517			at plugin module level and providing :meth:`register_argument`
518			method.
519
520			This class also provides a dictionary of widget types at its
521			:attr:`widget_types` attribute. They can be referenced by their keys on
522			both :meth:`create_widget` and :meth:`register_widget` methods' `type`
523			parameter, or instantiated directly and passed to :meth:`register_widget`
524			via `widget` parameter.
525			'''
526			def clear(self):
527			'''
528			Clear plugin manager state.
529
530			Registered widgets will be disposed after calling this method.
531
532			Registered mimetype functions will be disposed after calling this
533			method.
534
535			Registered command-line arguments will be disposed after calling this
536			method.
537			'''
538			super(PluginManager, self).clear()
539
540			@deprecated
541			def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
542			deprecated_places = {
543			'javascript': 'scripts',
544			'style': 'styles',
545			'button': 'entry-actions',
546			'link': 'entry-link',
547			None: 'none'
548			}
549			widget_mimetype_re = re.compile(
550			'^%s$' % '$\|^'.join(
551			map(re.escape, mimetypes)
552			).replace('\\*', '[^/]+')
553			)
554			print(widget_mimetype_re.pattern)
555			widget_type = getattr(widget, '_type', 'base')
556			widget_fields = self.widget_types[widget_type]._fields
557			widget_props = {
558			name: getattr(widget, name)
559			for name in widget_fields
560			if hasattr(widget, name)
561			}
562			widget_props.update(
563			type=widget_type,
564			place=deprecated_places.get(
565			widget_props['place'],
566			widget_props['place']
567			),
568			filter=lambda f: widget_mimetype_re.match(f.type) is not None
569			)
570			if 'endpoint' in widget_fields:
571			widget_props['endpoint'] = endpoint
572			self.register_widget(**widget_props)
573
574			@deprecated
575			def get_actions(self, file):
576			return self.get_widgets(file=file)
577

ergoithz / browsepy

Push — dev-4.1-unstable ( 90aa3b...e0128e )

PluginManager.get_actions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like