Source code for pluginmanager.module_manager
import os
import sys
import logging
import inspect
from .compat import load_source
from pluginmanager import util
logging.basicConfig()
[docs]class ModuleManager(object):
"""
`ModuleManager` manages the module plugin filter state and is responsible
for both loading the modules from source code and collecting the plugins
from each of the modules.
`ModuleManager` can also optionally manage modules explicitly through
the use of the add/get/set loaded modules methods. The default
implementation is hardwired to use the tracked loaded modules if no
modules are passed into the `collect_plugins` method.
"""
[docs] def __init__(self, module_plugin_filters=None):
"""
`module_plugin_filters` are callable filters. Each filter must take
in a list of plugins and a list of plugin names in the form of:
::
def my_module_plugin_filter(plugins: list, plugin_names: list):
pass
`module_plugin_filters` should return a list of the plugins and may
be either a single object or an iterable.
"""
if module_plugin_filters is None:
module_plugin_filters = []
module_plugin_filters = util.return_list(module_plugin_filters)
self.loaded_modules = set()
self.processed_filepaths = dict()
self.module_plugin_filters = module_plugin_filters
self._log = logging.getLogger(__name__)
self._error_string = 'pluginmanager unable to import {}\n'
[docs] def load_modules(self, filepaths):
"""
Loads the modules from their `filepaths`. A filepath may be
a directory filepath if there is an `__init__.py` file in the
directory.
If a filepath errors, the exception will be caught and logged
in the logger.
Returns a list of modules.
"""
# removes filepaths from processed if they are not in sys.modules
self._update_loaded_modules()
filepaths = util.return_set(filepaths)
modules = []
for filepath in filepaths:
filepath = self._clean_filepath(filepath)
# check to see if already processed and move onto next if so
if self._processed_filepath(filepath):
continue
module_name = util.get_module_name(filepath)
plugin_module_name = util.create_unique_module_name(module_name)
try:
module = load_source(plugin_module_name, filepath)
# Catch all exceptions b/c loader will return errors
# within the code itself, such as Syntax, NameErrors, etc.
except Exception:
exc_info = sys.exc_info()
self._log.error(msg=self._error_string.format(filepath),
exc_info=exc_info)
continue
self.loaded_modules.add(module.__name__)
modules.append(module)
self.processed_filepaths[module.__name__] = filepath
return modules
[docs] def collect_plugins(self, modules=None):
"""
Collects all the plugins from `modules`.
If modules is None, collects the plugins from the loaded modules.
All plugins are passed through the module filters, if any are any,
and returned as a list.
"""
if modules is None:
modules = self.get_loaded_modules()
else:
modules = util.return_list(modules)
plugins = []
for module in modules:
module_plugins = [(item[1], item[0])
for item
in inspect.getmembers(module)
if item[1] and item[0] != '__builtins__']
module_plugins, names = zip(*module_plugins)
module_plugins = self._filter_modules(module_plugins, names)
plugins.extend(module_plugins)
return plugins
[docs] def set_module_plugin_filters(self, module_plugin_filters):
"""
Sets the internal module filters to `module_plugin_filters`
`module_plugin_filters` may be a single object or an iterable.
Every module filters must be a callable and take in
a list of plugins and their associated names.
"""
module_plugin_filters = util.return_list(module_plugin_filters)
self.module_plugin_filters = module_plugin_filters
[docs] def add_module_plugin_filters(self, module_plugin_filters):
"""
Adds `module_plugin_filters` to the internal module filters.
May be a single object or an iterable.
Every module filters must be a callable and take in
a list of plugins and their associated names.
"""
module_plugin_filters = util.return_list(module_plugin_filters)
self.module_plugin_filters.extend(module_plugin_filters)
[docs] def get_module_plugin_filters(self, filter_function=None):
"""
Gets the internal module filters. Returns a list object.
If supplied, the `filter_function` should take in a single
list argument and return back a list. `filter_function` is
designed to given the option for a custom filter on the module filters.
"""
if filter_function is None:
return self.module_plugin_filters
else:
return filter_function(self.module_plugin_filters)
[docs] def remove_module_plugin_filters(self, module_plugin_filters):
"""
Removes `module_plugin_filters` from the internal module filters.
If the filters are not found in the internal representation,
the function passes on silently.
`module_plugin_filters` may be a single object or an iterable.
"""
util.remove_from_list(self.module_plugin_filters,
module_plugin_filters)
def _get_modules(self, names):
"""
An internal method that gets the `names` from sys.modules and returns
them as a list
"""
loaded_modules = []
for name in names:
loaded_modules.append(sys.modules[name])
return loaded_modules
[docs] def add_to_loaded_modules(self, modules):
"""
Manually add in `modules` to be tracked by the module manager.
`modules` may be a single object or an iterable.
"""
modules = util.return_set(modules)
for module in modules:
if not isinstance(module, str):
module = module.__name__
self.loaded_modules.add(module)
[docs] def get_loaded_modules(self):
"""
Returns all modules loaded by this instance.
"""
return self._get_modules(self.loaded_modules)
def _filter_modules(self, plugins, names):
"""
Internal helper method to parse all of the plugins and names
through each of the module filters
"""
if self.module_plugin_filters:
# check to make sure the number of plugins isn't changing
original_length_plugins = len(plugins)
module_plugins = set()
for module_filter in self.module_plugin_filters:
module_plugins.update(module_filter(plugins, names))
if len(plugins) < original_length_plugins:
warning = """Module Filter removing plugins from original
data member! Suggest creating a new list in each module
filter and returning new list instead of modifying the
original data member so subsequent module filters can have
access to all the possible plugins.\n {}"""
self._log.info(warning.format(module_filter))
plugins = module_plugins
return plugins
def _clean_filepath(self, filepath):
"""
processes the filepath by checking if it is a directory or not
and adding `.py` if not present.
"""
if (os.path.isdir(filepath) and
os.path.isfile(os.path.join(filepath, '__init__.py'))):
filepath = os.path.join(filepath, '__init__.py')
if (not filepath.endswith('.py') and
os.path.isfile(filepath + '.py')):
filepath += '.py'
return filepath
def _processed_filepath(self, filepath):
"""
checks to see if the filepath has already been processed
"""
processed = False
if filepath in self.processed_filepaths.values():
processed = True
return processed
def _update_loaded_modules(self):
"""
Updates the loaded modules by checking if they are still in sys.modules
"""
system_modules = sys.modules.keys()
for module in list(self.loaded_modules):
if module not in system_modules:
self.processed_filepaths.pop(module)
self.loaded_modules.remove(module)