PluginManagerBase.import_plugin() - Code Metrics - Inspection of "more doc and tests" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1-unstable ( 82d826...66ea16 )

by Felipe A.

created 2016-11-24 14:43 UTC

PluginManagerBase.import_plugin() D

↳ Parent: PluginManagerBase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
dl	0
loc	29
rs	4
c	0
b	0
f	0

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

import re
import sys
import argparse
import warnings
import collections

from flask import current_app
from werkzeug.utils import cached_property

from . import mimetype
from . import compat
from .compat import deprecated, usedoc


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 InvalidArgumentError(ValueError):
    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%s' % (namespace, '' if namespace[-1] == '_' else '.', 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 bool(widget) == bool(place or type):
            raise InvalidArgumentError(
                '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 MimetypeActionPluginManager(WidgetPluginManager, MimetypePluginManager):
    '''
    Deprecated plugin API
    '''

    _deprecated_places = {
        'javascript': 'scripts',
        'style': 'styles',
        'button': 'entry-actions',
        'link': 'entry-link',
        }

    @classmethod
    def _mimetype_filter(cls, mimetypes):
        widget_mimetype_re = re.compile(
            '^%s$' % '$|^'.join(
                map(re.escape, mimetypes)
                ).replace('\\*', '[^/]+')
            )

        def handler(f):
            return widget_mimetype_re.match(f.type) is not None

        return handler

    def _widget_attrgetter(self, widget, name):
        def handler(f):
            app = f.app or self.app or current_app
            with app.app_context():
                return getattr(widget.for_file(f), name)
        return handler

    def _widget_props(self, widget, endpoint=None, mimetypes=(),
                      dynamic=False):
        type = getattr(widget, '_type', 'base')
        fields = self.widget_types[type]._fields
        with self.app.app_context():
            props = {
                name: self._widget_attrgetter(widget, name)
                for name in fields
                if hasattr(widget, name)
                }
        props.update(
            type=type,
            place=self._deprecated_places.get(widget.place),
            )
        if dynamic:
            props['filter'] = self._mimetype_filter(mimetypes)
        if 'endpoint' in fields:
            props['endpoint'] = endpoint
        return props

    @usedoc(WidgetPluginManager.__init__)
    def __init__(self, app=None):
        self._action_widgets = []
        super(MimetypeActionPluginManager, self).__init__(app=app)

    @usedoc(WidgetPluginManager.clear)
    def clear(self):
        self._action_widgets[:] = ()
        super(MimetypeActionPluginManager, self).clear()

    @cached_property
    def _widget(self):
        with warnings.catch_warnings():
            warnings.filterwarnings('ignore', category=DeprecationWarning)
            from . import widget
        return widget

    @cached_property
    @deprecated('Deprecated attribute action_class')
    def action_class(self):
        return collections.namedtuple(
            'MimetypeAction',
            ('endpoint', 'widget')
            )

    @cached_property
    @deprecated('Deprecated attribute style_class')
    def style_class(self):
        return self._widget.StyleWidget

    @cached_property
    @deprecated('Deprecated attribute button_class')
    def button_class(self):
        return self._widget.ButtonWidget

    @cached_property
    @deprecated('Deprecated attribute javascript_class')
    def javascript_class(self):
        return self._widget.JavascriptWidget

    @cached_property
    @deprecated('Deprecated attribute link_class')
    def link_class(self):
        return self._widget.LinkWidget

    @deprecated('Deprecated method register_action')
    def register_action(self, endpoint, widget, mimetypes=(), **kwargs):
        props = self._widget_props(widget, endpoint, mimetypes, True)
        self.register_widget(**props)
        self._action_widgets.append((widget, props['filter'], endpoint))

    @deprecated('Deprecated method get_actions')
    def get_actions(self, file):
        return [
            self.action_class(endpoint, deprecated.for_file(file))
            for deprecated, filter, endpoint in self._action_widgets
            if endpoint and filter(file)
            ]

    @usedoc(WidgetPluginManager.register_widget)
    def register_widget(self, place=None, type=None, widget=None, filter=None,
                        **kwargs):
        if isinstance(place or widget, self._widget.WidgetBase):
            warnings.warn(
                'Deprecated use of register_widget',
                category=DeprecationWarning
                )
            widget = place or widget
            props = self._widget_props(widget)
            self.register_widget(**props)
            self._action_widgets.append((widget, None, None))
            return
        return super(MimetypeActionPluginManager, self).register_widget(
            place=place, type=type, widget=widget, filter=filter, **kwargs)

    @usedoc(WidgetPluginManager.get_widgets)
    def get_widgets(self, file=None, place=None):
        if isinstance(file, compat.basestring) or \
          place in self._deprecated_places:
            warnings.warn(
                'Deprecated use of get_widgets',
                category=DeprecationWarning
                )
            place = file or place
            return [
                widget
                for widget, filter, endpoint in self._action_widgets
                if not (filter or endpoint) and place == widget.place
                ]
        return super(MimetypeActionPluginManager, self).get_widgets(
            file=file, place=place)


class PluginManager(MimetypeActionPluginManager,
                    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()


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

ergoithz / browsepy

Push — dev-4.1-unstable ( 82d826...66ea16 )

PluginManagerBase.import_plugin() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like