Source code for cement.ext.ext_plugin

Cement plugin extension module.

import os
import sys
import importlib
import importlib.util
import importlib.machinery
import re
from ..core import plugin, exc
from ..utils.misc import is_true, minimal_logger
from ..utils.fs import abspath

LOG = minimal_logger(__name__)

# FIX ME: This is a redundant name... ?

[docs] class CementPluginHandler(plugin.PluginHandler): """ This class is an internal implementation of the :ref:`IPlugin <cement.core.plugin>` interface. It does not take any parameters on initialization. """
[docs] class Meta: """Handler meta-data.""" label = 'cement' """The string identifier for this class."""
def __init__(self): super().__init__() self._loaded_plugins = [] self._enabled_plugins = [] self._disabled_plugins = []
[docs] def _setup(self, app_obj): super()._setup(app_obj) self._enabled_plugins = [] self._disabled_plugins = [] self.bootstrap = self.load_dirs = # parse all app configs for plugins. Note: these are already # loaded from files when app.config was setup. The application # configuration OVERRIDES plugin configs. for section in if not section.startswith('plugin.'): continue plugin_section = section plugin = re.sub('^plugin.', '', section) if 'enabled' not in continue if is_true(, 'enabled')): LOG.debug("enabling plugin '%s' per application config" % plugin) if plugin not in self._enabled_plugins: self._enabled_plugins.append(plugin) # pragma: nocover if plugin in self._disabled_plugins: self._disabled_plugins.remove(plugin) # pragma: nocover else: LOG.debug("disabling plugin '%s' per application config" % plugin) if plugin not in self._disabled_plugins: self._disabled_plugins.append(plugin) # pragma: nocover if plugin in self._enabled_plugins: self._enabled_plugins.remove(plugin) # pragma: nocover
[docs] def _load_plugin_from_dir(self, plugin_name, plugin_dir): """ Load a plugin from a directory path rather than a python package within sys.path. This would either be ```` or ``myplugin/`` within the given ``plugin_dir``. Args: plugin_name (str): The name of the plugin. plugin_dir (str): The filesystem directory path where the plugin exists. """ LOG.debug("attempting to load '%s' from '%s'" % (plugin_name, plugin_dir)) if not os.path.exists(plugin_dir): LOG.debug("plugin directory '%s' does not exist." % plugin_dir) return False spec = importlib.machinery.PathFinder().find_spec( plugin_name, [plugin_dir] ) if not spec: LOG.debug("plugin '%s' does not exist in '%s'." % (plugin_name, plugin_dir)) return False # We don't catch this because it would make debugging a # nightmare mod = importlib.util.module_from_spec(spec) sys.modules[plugin_name] = mod spec.loader.exec_module(mod) if mod and hasattr(mod, 'load'): mod.load( return True
[docs] def _load_plugin_from_bootstrap(self, plugin_name, base_package): """ Load a plugin from a python package. Returns True if no ImportError is encountered. Args: plugin_name (str): The name of the plugin, also the name of the module to load from base_package. I.e. ``myapp.bootstrap.myplugin`` base_package: The base python package to load the plugin module from. I.e. ``myapp.bootstrap`` or similar. Returns: bool: ``True`` is the plugin was loaded, ``False`` otherwise Raises: :py:class:`ImportError`: If the plugin can not be imported """ full_module = '%s.%s' % (base_package, plugin_name) # If the base package doesn't exist, we return False rather than # bombing out. if base_package not in sys.modules: try: __import__(base_package, globals(), locals(), [], 0) except ImportError: LOG.debug("unable to import plugin bootstrap module '%s'." % base_package) return False LOG.debug("attempting to load '%s' from '%s'" % (plugin_name, base_package)) # We don't catch this because it would make debugging a nightmare # FIXME: not sure how to test/cover this if full_module not in sys.modules: __import__(full_module, globals(), locals(), [], 0) # pragma: nocover if hasattr(sys.modules[full_module], 'load'): sys.modules[full_module].load( return True
[docs] def load_plugin(self, plugin_name): """ Load a plugin whose name is ``plugin_name``. First attempt to load from a plugin directory (plugin_dir), secondly attempt to load from a Python module determined by ``App.Meta.plugin_module``. Upon successful loading of a plugin, the plugin name is appended to the ``self._loaded_plugins list``. Args: plugin_name (str): The name of the plugin to load. Raises: cement.core.exc.FrameworkError: If the plugin can not be loaded """ LOG.debug("loading application plugin '%s'" % plugin_name) # first attempt to load from plugin_dirs for load_dir in self.load_dirs: load_dir = abspath(load_dir) if self._load_plugin_from_dir(plugin_name, load_dir): self._loaded_plugins.append(plugin_name) break # then from a bootstrap module if plugin_name not in self._loaded_plugins: if self._load_plugin_from_bootstrap(plugin_name, self.bootstrap): self._loaded_plugins.append(plugin_name) # otherwise it's a bust if plugin_name not in self._loaded_plugins: raise exc.FrameworkError("Unable to load plugin '%s'." % plugin_name)
[docs] def load_plugins(self, plugin_list): """ Load a list of plugins. Each plugin name is passed to ``self.load_plugin()``. Args: plugin_list (list): A list of plugin names to load. """ for plugin_name in plugin_list: self.load_plugin(plugin_name)
[docs] def get_loaded_plugins(self): """List of plugins that have been loaded.""" return self._loaded_plugins
[docs] def get_enabled_plugins(self): """List of plugins that are enabled (not necessary loaded yet).""" return self._enabled_plugins
[docs] def get_disabled_plugins(self): """List of disabled plugins""" return self._disabled_plugins
def load(app): app.handler.register(CementPluginHandler)