# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright 2015-2018 by Exopy Authors, see AUTHORS for more details.
#
# Distributed under the terms of the BSD license.
#
# The full license is in the file LICENCE, distributed with this software.
# -----------------------------------------------------------------------------
"""Useful tools to avoid code duplication when writing plugins.
"""
import sys
from collections import defaultdict
from atom.api import Atom, Dict, Str, Coerced, Typed, Callable, List
from enaml.workbench.api import Workbench, Plugin
from .atom_util import (update_members_from_preferences,
preferences_from_members)
[docs]class HasPreferencesPlugin(Plugin):
""" Base class for plugin using preferences.
Simply defines the most basic preferences system inherited from
HasPrefAtom. Preferences are automatically queried and saved using the
exopy.app.preferences plugin.
"""
update_members_from_preferences = update_members_from_preferences
preferences_from_members = preferences_from_members
[docs] def start(self):
"""Upon starting initialize members using preferences.
"""
core = self.workbench.get_plugin('enaml.workbench.core')
prefs = core.invoke_command('exopy.app.preferences.get',
{'plugin_id': self.manifest.id})
self.update_members_from_preferences(prefs)
core.invoke_command('exopy.app.preferences.plugin_init_complete',
{'plugin_id': self.manifest.id})
[docs]def make_handler(id, method_name):
"""Generate a generic handler calling a plugin method.
"""
def handler(event):
"""Handler getting the method corresponding to the command from the
plugin.
"""
pl = event.workbench.get_plugin(id)
return getattr(pl, method_name)(**event.parameters)
handler.__name__ += '_' + method_name
return handler
[docs]def make_extension_validator(base_cls, fn_names=(),
attributes=('description',)):
"""Create an extension validation function checking that key methods were
overridden and attributes values provided.
Parameters
----------
base_cls : type
Base class from which the contribution should inherit.
fn_names : iterable[unicode], optional
Names of the function the extensions must override.
attributes : iterable[unicode], optional
Names of the attributes the extension should provide values for.
Returns
-------
validator : callable
Function that can be used to validate an extension contribution.
"""
def validator(contrib):
"""Validate the children of an extension.
"""
for name in fn_names:
member = getattr(contrib, name)
# Compatibilty trick for Enaml declarative function (not necessary
# for enaml compatible with Python 3)
func = getattr(member, 'im_func',
getattr(member, '__func__', None))
o_func = getattr(base_cls, name)
if not func or func is o_func:
msg = "%s '%s' does not declare a %s function"
return False, msg % (base_cls, contrib.id, name)
for attr in attributes:
if not getattr(contrib, attr):
msg = '%s %s does not provide a %s'
return False, msg % (base_cls, contrib.id, attr)
return True, ''
doc = 'Ensure that %s subclasses does override %s' % (base_cls, fn_names)
validator.__doc__ = doc
return validator
[docs]class ClassTuple(tuple):
"""Special tuple meant to hold classes.
Provides an smart constructor and a nice str representation.
"""
def __new__(cls, vals):
if isinstance(vals, type):
return tuple.__new__(ClassTuple, [vals])
else:
return tuple.__new__(ClassTuple, vals)
def __str__(self):
return ', '.join((c.__name__ for c in self))
[docs]class BaseCollector(Atom):
"""Base class for automating extension collection.
"""
#: Reference to the application workbench.
workbench = Typed(Workbench)
#: Id of the extension point to observe.
point = Str()
#: Expected class(es) of the object generated by the extension.
ext_class = Coerced(ClassTuple)
#: Dictionary storing the consributiosn of the observed extension point.
#: This should not be altered by user code. This is never modified in place
#: so user code will get reliable notifications when observing it.
contributions = Dict()
[docs] def start(self):
"""Run first collections of contributions and set up observers.
This method should be called in the start method of the plugin using
this object.
"""
self._refresh_contributions()
self._bind_observers()
[docs] def stop(self):
"""Unbind observers and clean up ressources.
This method should be called in the stop method of the plugin using
this object.
"""
self._unbind_observers()
self.unobserve('contributions') # Dicsonnect all observers
self.contributions.clear()
self._extensions.clear()
# =========================================================================
# --- Private API ---------------------------------------------------------
# =========================================================================
#: Private storage keeping track of which extension declared which object.
_extensions = Typed(defaultdict, (list,))
def _refresh_contributions(self):
""" Refresh the extensions contributions.
This method should be called in the start method of the plugin using
this object.
"""
raise NotImplementedError()
def _on_contribs_updated(self, change):
""" The observer for the extension point
"""
raise NotImplementedError()
def _bind_observers(self):
""" Setup the observers for the extension point.
This method should be called in the start method of the plugin using
this object.
"""
workbench = self.workbench
point = workbench.get_extension_point(self.point)
point.observe('extensions', self._on_contribs_updated)
def _unbind_observers(self):
""" Remove the observers for the plugin.
This method should be called in the stop method of the plugin using
this object.
"""
workbench = self.workbench
point = workbench.get_extension_point(self.point)
point.unobserve('extensions', self._on_contribs_updated)
[docs]class ExtensionsCollector(BaseCollector):
"""Convenience class collecting an extension point contribution.
This class can be used on any extension point to which extensions
contribute instances of a specific class. Those object should always have
an id member.
"""
#: Callable to use to ensure that the provide extension does fit.
#: Should take the proposed contribution as single argument and return a
#: bool indicating the result of the test, and a message explaining what
#: went wrong (or an empty string if test passed).
validate_ext = Callable(lambda e: (True, ''))
[docs] def contributed_by(self, contrib_id):
"""Find the extension declaring a contribution.
"""
contrib = self.contributions[contrib_id]
for ext, cs in self._extensions.items():
if contrib in cs:
return ext
# =========================================================================
# --- Private API ---------------------------------------------------------
# =========================================================================
#: Private storage keeping track of which extension declared which object.
_extensions = Typed(defaultdict, (list,))
def _refresh_contributions(self):
""" Refresh the extensions contributions.
This method should be called in the start method of the plugin using
this object.
"""
tb = {}
workbench = self.workbench
point = workbench.get_extension_point(self.point)
# If no extension remain clear everything
if not point or not point.extensions:
# Force a notification to be emitted.
self.contributions = {}
self._extensions.clear()
return
extensions = point.extensions
# Get the contributions declarations for all extensions.
new_extensions = defaultdict(list)
old_extensions = self._extensions
for extension in extensions:
if extension in old_extensions:
contribs = old_extensions[extension]
else:
try:
contribs = self._load_contributions(extension)
except TypeError as e:
tb['Extension ' + extension.qualified_id] = '{}'.format(e)
continue
new_extensions[extension].extend(contribs)
# Create mapping between contrib id and declaration.
contribs = {}
for extension in extensions:
for contrib in new_extensions[extension]:
if contrib.id in contribs:
msg = "{} attempted to register already registered '{}'"
pattern = 'Duplicate ' + contrib.id + '_{}'
i = 0
while pattern.format(i) in tb:
i += 1
tb[pattern.format(i)] = \
msg.format(extension.qualified_id, contrib.id)
res, msg = self.validate_ext(contrib)
if not res:
ext = 'While loading {},'.format(extension.qualified_id)
tb[contrib.id] = ext + msg
contribs[contrib.id] = contrib
self.contributions = contribs
self._extensions = new_extensions
if tb:
core = self.workbench.get_plugin('enaml.workbench.core')
core.invoke_command('exopy.app.errors.signal',
{'kind': 'extensions', 'point': self.point,
'errors': tb})
def _load_contributions(self, extension):
""" Load the contributed objects for the given extension.
Parameters
----------
extension : Extension
The extension object of interest.
Returns
-------
contribs : list
The objects declared by the extension.
"""
workbench = self.workbench
contribs = extension.get_children(self.ext_class)
if extension.factory is not None and not contribs:
for contrib in extension.factory(workbench):
if not isinstance(contrib, self.ext_class):
msg = "extension '{}' created non-{}."
raise TypeError(msg.format(extension.qualified_id,
self.ext_class))
contribs.append(contrib)
return contribs
def _on_contribs_updated(self, change):
""" The observer for the extension point
"""
self._refresh_contributions()
[docs]class DeclaratorsCollector(BaseCollector):
"""Class registering Declarator contributed to an extension point.
This class can be used on any extension point to which extensions
contribute Declarator.
"""
# =========================================================================
# --- Private API ---------------------------------------------------------
# =========================================================================
#: Temporary list in which declarations which cannot yet be taken into
#: account because another declaration has not yet been registered.
_delayed = List()
def _refresh_contributions(self):
"""Load all extensions contributed to the observed point.
"""
workbench = self.workbench
point = workbench.get_extension_point(self.point)
extensions = point.extensions
self._register_decls(extensions)
def _register_decls(self, extensions):
"""Register the declaration linked to some extensions.
Handle multiple registering attempts.
"""
# Get the declarators for all extensions.
tb = {}
contributions = self.contributions.copy()
new_extensions = defaultdict(list)
old_extensions = self._extensions
for extension in extensions:
if extension not in old_extensions:
try:
declarators = self._get_decls(extension)
except TypeError as e:
tb['Extension ' + extension.qualified_id] = '{}'.format(e)
continue
new_extensions[extension].extend(declarators)
# Register all contributions.
for extension in new_extensions:
for declarator in new_extensions[extension]:
declarator.register(self, tb)
# Handle delayed registering.
old = 0
while old != len(self._delayed) and self._delayed:
# Copy and clean delayed list.
delayed = self._delayed[:]
old = len(delayed)
self._delayed = []
# Attempt to re-register delayed declarators
for declarator in delayed:
declarator.register(self, tb)
if self._delayed:
msg = 'Some declarations have not been registered : {}'
tb['Missing declarations'] = msg.format(self._delayed)
self._extensions.update(new_extensions)
if self.contributions != contributions:
c = self.contributions
with self.suppress_notifications():
self.contributions = contributions
self.contributions = c
if tb:
core = self.workbench.get_plugin('enaml.workbench.core')
core.invoke_command('exopy.app.errors.signal',
{'kind': 'extensions', 'point': self.point,
'errors': tb})
def _get_decls(self, extension):
"""Get the task declarations declared by an extension.
"""
workbench = self.workbench
contribs = extension.get_children(self.ext_class)
if extension.factory is not None and not contribs:
for contrib in extension.factory(workbench):
if not isinstance(contrib, self.ext_class):
msg = "Extension '{}' should create {} not {}."
raise TypeError(msg.format(extension.qualified_id,
self.ext_class,
type(contrib).__name__))
contribs.append(contrib)
return contribs
def _unregister_decls(self, extensions):
"""Unregister the declarations linked to some extensions.
"""
contributions = self.contributions.copy()
for extension in extensions:
for declarator in extensions[extension]:
declarator.unregister(self)
del self._extensions[extension]
if self.contributions != contributions:
c = self.contributions
with self.suppress_notifications():
self.contributions = contributions
self.contributions = c
def _on_contribs_updated(self, change):
"""Update the registered declarations when an extension is
added/removed.
"""
old = set(change.get('oldvalue', ()))
new = set(change['value'])
# If no extension remain clear everything
if not new:
# Force a notification to be emitted.
self.contributions = {}
self._extensions.clear()
return
added = new - old
removed = old - new
self._unregister_decls({ext: d for ext, d in self._extensions.items()
if ext in removed})
self._register_decls(added)