PluginManagerBase.import_plugin() - Code Metrics - Inspection of "document plugin manager" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1-unstable ( 38bef3...b6e12b )

by Felipe A.

created 2016-11-22 12:26 UTC

PluginManagerBase.import_plugin() C

↳ Parent: PluginManagerBase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

import sys
import argparse
import warnings
import collections

from . import mimetype


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'),
            {
                'text': lambda f: f.name,
                'icon': lambda f: f.category
            }),
        'button': defaultsnamedtuple(
            'Button',
            ('place', 'type', 'css', 'text', 'endpoint', 'href')),
        'upload': defaultsnamedtuple(
            'Upload',
            ('place', 'type', 'css', 'text', 'endpoint', 'action')),
        'stylesheet': defaultsnamedtuple(
            'Stylesheet',
            ('place', 'type', 'endpoint', 'filename', 'href')),
        'script': defaultsnamedtuple(
            'Script',
            ('place', 'type', 'endpoint', 'filename', 'src')),
        'html': defaultsnamedtuple(
            'Html',
            ('place', 'type', 'html')),
    }

    def clear(self):
        '''
        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))

    def _resolve_widget(self, 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()


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

ergoithz / browsepy

Push — dev-4.1-unstable ( 38bef3...b6e12b )

PluginManagerBase.import_plugin() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like