Source code for envisage.composite_plugin_manager

# (C) Copyright 2007-2024 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!
""" A plugin manager composed of other plugin managers! """


# Standard library imports.
import logging

# Enthought library imports.
from traits.api import (
    Event,
    HasTraits,
    Instance,
    List,
    observe,
    on_trait_change,
    provides,
)

# Local imports.
from .i_application import IApplication
from .i_plugin import IPlugin
from .i_plugin_manager import IPluginManager
from .plugin_event import PluginEvent
from .plugin_manager import PluginManager

# Logging.
logger = logging.getLogger(__name__)


[docs]@provides(IPluginManager)
class CompositePluginManager(HasTraits):
    """A plugin manager composed of other plugin managers!

    e.g::

        plugin_manager = CompositePluginManager(
             plugin_managers = [
                 EggBasketPluginManager(...),
                 PackagePluginManager(...),
             ]
        )

    """

    #### 'IPluginManager' protocol ############################################

    #### Events ####

    # Fired when a plugin has been added to the manager.
    plugin_added = Event(PluginEvent)

    # Fired when a plugin has been removed from the manager.
    plugin_removed = Event(PluginEvent)

    #### 'CompositePluginManager' protocol ####################################

    # The application that the plugin manager is part of.
    application = Instance(IApplication)

    @observe("application")
    def _update_application(self, event):
        new = event.new
        for plugin_manager in self.plugin_managers:
            plugin_manager.application = new

    # The plugin managers that make up this plugin manager!
    #
    # This is currently a list of 'PluginManager's as opposed to, the more
    # preferable 'IPluginManager' because the interface doesn't currently
    # have an 'application' trait. Should we move 'application' up to
    # 'IPluginManager'?
    plugin_managers = List(PluginManager)

    @on_trait_change("plugin_managers[]")
    def _update_application_on_plugins(self, obj, trait_named, removed, added):
        for plugin_manager in removed:
            plugin_manager.application = self.application

        for plugin_manager in added:
            plugin_manager.application = self.application

    @on_trait_change("plugin_managers:plugin_added")
    def _plugin_added(self, obj, trait_name, old, new):
        self.plugin_added = new

    @on_trait_change("plugin_managers:plugin_removed")
    def _plugin_removed(self, obj, trait_name, old, new):
        self.plugin_removed = new

    #### Private protocol #####################################################

    # The plugins that the manager manages!
    _plugins = List(IPlugin)

    def __plugins_default(self):
        plugins = []
        for plugin_manager in self.plugin_managers:
            for plugin in plugin_manager:
                plugins.append(plugin)

        return plugins

    #### 'object' protocol ####################################################

    def __iter__(self):
        """Return an iterator over the manager's plugins."""

        plugins = []
        for plugin_manager in self.plugin_managers:
            for plugin in plugin_manager:
                plugins.append(plugin)

        return iter(plugins)

    #### 'IPluginManager' protocol ############################################

[docs]    def add_plugin(self, plugin):
        """Add a plugin to the manager."""

        raise NotImplementedError

[docs]    def get_plugin(self, plugin_id):
        """Return the plugin with the specified Id."""

        for plugin in self:
            if plugin_id == plugin.id:
                break

        else:
            plugin = None

        return plugin

[docs]    def remove_plugin(self, plugin):
        """Remove a plugin from the manager."""

        raise NotImplementedError

[docs]    def start(self):
        """Start the plugin manager."""

        for plugin in self:
            self.start_plugin(plugin)

[docs]    def start_plugin(self, plugin=None, plugin_id=None):
        """Start the specified plugin."""

        plugin = plugin or self.get_plugin(plugin_id)
        if plugin is not None:
            logger.debug("plugin %s starting", plugin.id)
            plugin.activator.start_plugin(plugin)
            logger.debug("plugin %s started", plugin.id)

        else:
            raise ValueError("no such plugin %s" % plugin_id)

[docs]    def stop(self):
        """Stop the plugin manager."""

        # We stop the plugins in the reverse order that they were started.
        stop_order = list(iter(self))
        stop_order.reverse()

        for plugin in stop_order:
            self.stop_plugin(plugin)

[docs]    def stop_plugin(self, plugin=None, plugin_id=None):
        """Stop the specified plugin."""

        plugin = plugin or self.get_plugin(plugin_id)
        if plugin is not None:
            logger.debug("plugin %s stopping", plugin.id)
            plugin.activator.stop_plugin(plugin)
            logger.debug("plugin %s stopped", plugin.id)

        else:
            raise ValueError("no such plugin %s" % plugin_id)
Source code for envisage.composite_plugin_manager

Quick search
