<s> import functools <EOL> import logging <EOL> import pkgutil <EOL> import re <EOL> import warnings <EOL> from twisted import plugin as twisted_plugin <EOL> from twisted . python import reflect , failure , log as twisted_log , zippath <EOL> from piped import event , exceptions , util <EOL> logger = logging . getLogger ( __name__ ) <EOL> def _plugin_error_handler ( package ) : <EOL> f = failure . Failure ( ) <EOL> f . trap ( ImportError ) <EOL> log_func = logger . error <EOL> if util . in_unittest ( ) : <EOL> log_func = logger . info <EOL> log_func ( '<STR_LIT>' % ( package . __name__ , f . type . __name__ , f . getErrorMessage ( ) ) ) <EOL> def getPlugins ( interface , package ) : <EOL> original_err = twisted_log . err <EOL> try : <EOL> twisted_log . err = functools . partial ( _plugin_error_handler , package ) <EOL> for plugin in twisted_plugin . getPlugins ( interface , package ) : <EOL> yield plugin <EOL> finally : <EOL> twisted_log . err = original_err <EOL> class PluginManager ( object ) : <EOL> plugin_interface = None <EOL> plugin_packages = None <EOL> plugin_configuration_path = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> self . _plugins = set ( ) <EOL> self . _plugin_factory_by_name = dict ( ) <EOL> self . _providers_by_keyword = dict ( ) <EOL> self . runtime_environment = None <EOL> self . plugins_loaded = False <EOL> self . on_plugins_loaded = event . Event ( ) <EOL> self . on_package_paths_updated = event . Event ( ) <EOL> def configure ( self , runtime_environment ) : <EOL> self . runtime_environment = runtime_environment <EOL> self . plugins_config = runtime_environment . get_configuration_value ( self . plugin_configuration_path , dict ( ) ) <EOL> self . load_plugins ( ) <EOL> def load_plugins ( self , reload = False ) : <EOL> """<STR_LIT>""" <EOL> monkey_patched_zippath = False <EOL> if not hasattr ( zippath . ZipPath , '<STR_LIT>' ) : <EOL> monkey_patched_zippath = True <EOL> zippath . ZipPath . setContent = lambda * a , ** kw : None <EOL> try : <EOL> if not self . plugins_loaded or reload : <EOL> self . _plugins = set ( ) <EOL> for package in self . plugin_packages : <EOL> for plugin in self . _get_plugins_from_package ( package ) : <EOL> self . _plugins . add ( plugin ) <EOL> for package in self . _get_bundled_packages ( ) : <EOL> for plugin in self . _get_plugins_from_package ( package ) : <EOL> self . _plugins . add ( plugin ) <EOL> self . _plugin_factory_by_name = dict ( ) <EOL> for plugin in self . _plugins : <EOL> self . _register_plugin ( plugin ) <EOL> self . plugins_loaded = True <EOL> finally : <EOL> if monkey_patched_zippath : <EOL> delattr ( zippath . ZipPath , '<STR_LIT>' ) <EOL> self . on_plugins_loaded ( self . _plugins ) <EOL> def _get_bundled_packages ( self ) : <EOL> """<STR_LIT>""" <EOL> packages = list ( ) <EOL> for bundle_name , bundle_config in self . plugins_config . get ( '<STR_LIT>' , dict ( ) ) . items ( ) : <EOL> self . _fail_if_bundle_configuration_is_invalid ( bundle_config ) <EOL> for package_name in bundle_config : <EOL> try : <EOL> package = reflect . namedAny ( package_name ) <EOL> except reflect . InvalidName , e : <EOL> e_msg = e . args [ <NUM_LIT:0> ] <EOL> detail = '<STR_LIT>' % package_name <EOL> hint = '<STR_LIT>' <EOL> raise exceptions . ConfigurationError ( e_msg , detail , hint ) <EOL> packages . append ( package ) <EOL> return packages <EOL> def _is_disabled ( self , package_name ) : <EOL> for disabled_pattern in self . plugins_config . get ( '<STR_LIT>' , list ( ) ) : <EOL> if re . match ( disabled_pattern , package_name ) : <EOL> return True <EOL> return False <EOL> def _get_plugins_from_package ( self , package ) : <EOL> self . _fail_if_not_a_package ( package ) <EOL> if self . _is_disabled ( package . __name__ ) : <EOL> return <EOL> sub_packages_names = [ name for ( loader , name , ispkg ) in <EOL> pkgutil . walk_packages ( package . __path__ , package . __name__ + '<STR_LIT:.>' ) <EOL> if ispkg and not self . _is_disabled ( name ) ] <EOL> for sub_package_name in sub_packages_names : <EOL> for plugin in getPlugins ( self . plugin_interface , reflect . namedAny ( sub_package_name ) ) : <EOL> yield plugin <EOL> for plugin in twisted_plugin . getPlugins ( self . plugin_interface , package ) : <EOL> yield plugin <EOL> def _fail_if_not_a_package ( self , package ) : <EOL> if not hasattr ( package , '<STR_LIT>' ) : <EOL> e_msg = '<STR_LIT>' <EOL> detail = '<STR_LIT>' <EOL> module_name = package . __name__ <EOL> suggested_package_name = module_name . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> hint = '<STR_LIT>' % ( module_name , suggested_package_name ) <EOL> raise exceptions . ConfigurationError ( e_msg , detail , hint ) <EOL> def _fail_if_bundle_configuration_is_invalid ( self , bundle_config ) : <EOL> if not isinstance ( bundle_config , list ) : <EOL> e_msg = '<STR_LIT>' <EOL> detail = '<STR_LIT>' % bundle_config <EOL> raise exceptions . ConfigurationError ( e_msg , detail ) <EOL> def _register_plugin ( self , plugin ) : <EOL> name = getattr ( plugin , '<STR_LIT:name>' , None ) or reflect . fullyQualifiedName ( plugin ) <EOL> self . _fail_if_plugin_name_is_already_registered ( plugin , name ) <EOL> self . _plugin_factory_by_name [ name ] = plugin <EOL> provided_keywords = getattr ( plugin , '<STR_LIT>' , [ ] ) <EOL> for keyword in provided_keywords : <EOL> self . _providers_by_keyword . setdefault ( keyword , [ ] ) . append ( name ) <EOL> def _fail_if_plugin_name_is_already_registered ( self , plugin , plugin_name ) : <EOL> if plugin_name in self . _plugin_factory_by_name : <EOL> e_msg = '<STR_LIT>' % plugin_name <EOL> details = ( plugin , self . _plugin_factory_by_name [ plugin_name ] ) <EOL> detail = '<STR_LIT>' % details <EOL> raise exceptions . ConfigurationError ( e_msg , detail ) <EOL> def get_plugin_factory ( self , plugin_name ) : <EOL> """<STR_LIT>""" <EOL> plugin_factory = self . _plugin_factory_by_name . get ( plugin_name ) <EOL> if not plugin_factory : <EOL> self . _fail_because_of_nonexisting_plugin ( plugin_name ) <EOL> return plugin_factory <EOL> def _fail_because_of_nonexisting_plugin ( self , plugin_name ) : <EOL> if not self . plugins_loaded : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> e_msg = '<STR_LIT>' % ( plugin_name , ) <EOL> details = dict ( <EOL> available_plugins = sorted ( self . _plugin_factory_by_name . keys ( ) ) , <EOL> interface = self . plugin_interface . __name__ , <EOL> packages = sorted ( [ package . __name__ for package in self . plugin_packages + self . _get_bundled_packages ( ) ] ) , <EOL> ) <EOL> detail = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % details ) <EOL> hint = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise exceptions . ConfigurationError ( e_msg , detail , hint ) <EOL> def get_providers_of_keyword ( self , keyword ) : <EOL> """<STR_LIT>""" <EOL> return self . _providers_by_keyword . get ( keyword , [ ] ) <EOL> def get_all_plugins ( self ) : <EOL> """<STR_LIT>""" <EOL> return tuple ( self . _plugin_factory_by_name . items ( ) ) </s>
