#!/usr/bin/env python # # COPYRIGHT # # Permission to use, copy, modify, and distribute this software and # its documentation for any purpose and without fee is hereby # granted, provided that the above copyright notice appear in all # copies and that both that copyright notice and this permission # notice appear in supporting documentation, and that the name of Doug # Hellmann not be used in advertising or publicity pertaining to # distribution of the software without specific, written prior # permission. # # DISCLAIMER # # DOUG HELLMANN DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, # INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN # NO EVENT SHALL DOUG HELLMANN BE LIABLE FOR ANY SPECIAL, INDIRECT OR # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS # OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. # """Define a class to handle pluggable module loading. """ __rcs_info__ = { # # Creation Information # 'module_name':'$RCSfile: pluginloader.py,v $', 'creator':'Doug Hellmann ', 'project':'HappyDoc', 'created':'Sat, 03-Jun-2000 19:23:48 EDT', # # Current Information # 'author':'$Author: doughellmann $', 'version':'$Revision: 1.5 $', 'date':'$Date: 2003/01/18 19:56:24 $', } try: __version__ = __rcs_info__['version'].split(' ')[1] except: __version__ = '0.0' # # Import system modules # import glob import os import sys try: from cStringIO import StringIO except: from StringIO import StringIO import traceback import UserDict # # Import Local modules # from happydoclib.trace import trace as TRACE # # Module # TRACE_LEVEL=4 class PluginException(Exception): """Exception raised when any error occurs while loading a plugin. """ def __init__(self, wrappedException, pluginName): self._plugin_name = pluginName self._wrapped_exception = wrappedException return def __str__(self): return 'Problem with %s\n%s' % (self._plugin_name, self._wrapped_exception) # # Where is the application installed? # APP_HOME_DIR=os.path.dirname( sys.argv[0] ) sys.path.append(APP_HOME_DIR) class PluginLoader(UserDict.UserDict): """A class to handle pluggable module loading.""" def __init__(self, moduleName, modulePath, parentModulePrefix): """Create a PluginLoader. Parameters moduleName -- The imported module, from which we get name, path, etc. to find the sub-modules. modulePath -- The path to the module, and therefore the parent directory of the plugins. parentModulePrefix -- The prefix of parent names to be added to moduleName. For example 'happydoclib'. """ TRACE.into('PluginLoader', '__init__', moduleName=moduleName, modulePath=modulePath, parentModulePrefix=parentModulePrefix, outputLevel=TRACE_LEVEL, ) # # Initialize the base class # UserDict.UserDict.__init__(self) # # Name for these plugins # module_name_parts = moduleName.split('.') if len(module_name_parts) > 1: moduleName = '.'.join( module_name_parts[1:] ) self.plugin_set_name = moduleName #print '\n*** Initializing plugin set %s' % self.plugin_set_name # # Where the plugins will be # self.plugin_dir = modulePath #print ' From %s' % self.plugin_dir # # Name to add before importing # self.parent_module_prefix = parentModulePrefix # # List of plugins # _module_list = self.getModuleList() _module_list.sort() # # Load the modules # for _module_name in _module_list: TRACE.into('PluginLoader', '__init__:Loop', _module_name=_module_name, outputLevel=TRACE_LEVEL, ) module_name_parts = _module_name.split(os.sep) module_base_name = module_name_parts[-1] plugin_set_name = module_name_parts[-2] if plugin_set_name != self.plugin_set_name: print 'EEEE Got local plugin set name "%s" instead of "%s"' % \ (plugin_set_name, self.plugin_set_name) parent_package_name = module_name_parts[-3] TRACE.writeVar(parent_package_name=parent_package_name, outputLevel=TRACE_LEVEL, ) _module_name = os.path.basename(_module_name) if self.parent_module_prefix: _import_name = '%s.%s.%s' % (self.parent_module_prefix, plugin_set_name, module_base_name) else: _import_name = '%s.%s' % (plugin_set_name, module_base_name) TRACE.writeVar(_import_name=_import_name, outputLevel=TRACE_LEVEL, ) try: TRACE.write('Importing %s' % _import_name, outputLevel=TRACE_LEVEL, ) _module = __import__( _import_name ) TRACE.write('Done with import', outputLevel=TRACE_LEVEL) except Exception, msg: if int(os.environ.get('PLUGIN_DEBUG', '0')): buffer = StringIO() buffer.write('\n--- Plugin Module Error in %s ---\n' % _import_name) traceback.print_exc(file=buffer) TRACE.into('PluginLoader' 'Plugin module error', _import_name=_import_name, outputLevel=TRACE_LEVEL, ) TRACE.write(buffer.getvalue(), outputLevel=TRACE_LEVEL, ) TRACE.outof(outputLevel=TRACE_LEVEL) buffer.write('---------------------------\n\n') TRACE.outof(outputLevel=TRACE_LEVEL) raise PluginException( buffer.getvalue(), _import_name ) elif int(os.environ.get('PLUGIN_SILENT_ERRORS', '0')): TRACE.outof(outputLevel=TRACE_LEVEL) continue else: sys.stderr.write('\n--- Plugin Module Error in %s ---\n' % _import_name) traceback.print_exc() sys.stderr.write('---------------------------\n\n') TRACE.outof(outputLevel=TRACE_LEVEL) continue import_name_parts = _import_name.split('.') for sub_module in import_name_parts[1:]: try: _module = getattr(_module, sub_module) except AttributeError: sys.stderr.write('ERROR: Could not retrieve %s from %s\n' % \ (sub_module, _module.__name__)) raise try: info = _module.entryPoint() except AttributeError: sys.stderr.write('ERROR: Could not call entryPoint() from %s\n' % _import_name) sys.stderr.write('%s\n' % dir(_module)) raise else: self.addEntryPoint(info) TRACE.outof(outputLevel=TRACE_LEVEL) TRACE.outof(outputLevel=TRACE_LEVEL) return def getModuleList(self): "Return a list of module names to be used as plugins." glob_pattern = os.path.join( self.plugin_dir, '%s_*' % self.plugin_set_name, ) all_files = glob.glob(glob_pattern) all_basenames = [ os.path.splitext(x)[0] for x in all_files] unique_modules = [] for module in all_basenames: if module not in unique_modules: unique_modules.append(module) return unique_modules def addEntryPoint(self, infoDict): """Add an entry point into a module to our lookup tables. This method must be implemented by the subclass. """ raise 'Not implemented for %s' % self.__class__.__name__ if __name__ == '__main__': import os os.chdir('TestCases/test_plugin_loader') import sys sys.path.append( os.getcwd() ) import runtest