[docs]@publicclassPluginProvider():""" A base class for plugin locator """
[docs]defprovide(self):""" Should return a list of found plugin paths :returns: list(str) """raiseNotImplementedError()
[docs]@publicclassDirectoryPluginProvider(PluginProvider):""" A plugin provider that looks up plugins in a given directory. :param path: directory to look for plugins in """def__init__(self,path):self.path=os.path.abspath(path)
[docs]@publicclassPythonPathPluginProvider(PluginProvider):""" A plugin provider that looks up plugins on ``$PYTHONPATH`` """def__init__(self):pass
[docs]defprovide(self):found_plugins=[]forpathinsys.path:ifos.path.isdir(path):logging.debug(f'Looking for plugins in {path}')found_plugins+=DirectoryPluginProvider(path).provide()returnfound_plugins
[docs]defload_all_from(self,providers):""" Loads all plugins provided by given providers. :param providers: :type providers: list(:class:`PluginProvider`) """found=[]forproviderinproviders:found+=provider.provide()self.__plugin_info={}forpathinfound:try:yml_info=yaml.load(open(os.path.join(path,'plugin.yml')),Loader=yaml.SafeLoader)exceptyaml.constructor.ConstructorError:logging.error(f"Malformatted plugin.yml located at {path}, not loading plugin. ")continueyml_info['resources']=[({'path':x}ifisinstance(x,str)elsex)forxinyml_info.get('resources',[])]self.__plugin_info[yml_info['name']]={'info':yml_info,'path':path,'imported':False,}logging.info(f'Discovered {len(self.__plugin_info):d} plugins')self.load_order=[]to_load=list(self.__plugin_info.values())whileTrue:delta=0forplugininto_load:fordepinplugin['info']['dependencies']:ifisinstance(dep,PluginDependency)anddep.plugin_namenotinself.load_order:breakifisinstance(dep,OptionalPluginDependency)anddep.plugin_namenotinself.load_order:breakelse:self.load_order.append(plugin['info']['name'])to_load.remove(plugin)delta+=1ifdelta==0:# less strictforplugininto_load:fordepinplugin['info']['dependencies']:ifisinstance(dep,PluginDependency)anddep.plugin_namenotinself.load_order:breakelse:self.load_order.append(plugin['info']['name'])to_load.remove(plugin)delta+=1ifdelta==0:breakforplugininto_load:fordepinplugin['info']['dependencies']:ifisinstance(dep,PluginDependency)anddep.plugin_namenotinself.load_order:self.__crashes[plugin['info']['name']]=dep.build_exception()logging.warning(f"Not loading [{plugin['info']['name']}] because [{dep.plugin_name}] is unavailable")fornameinlist(self.load_order):try:fordependencyinself[name]['info']['dependencies']:ifnotisinstance(dependency,PluginDependency):dependency.check()exceptDependency.Unsatisfiedase:self.__crashes[name]=elogging.warning(f'Not loading [{name}] because dependency failed: {e}')self.load_order.remove(name)logging.debug(f'Resolved load order for {len(self.load_order):d} plugins: {self.load_order}')fornameinlist(self.load_order):try:self.__import_plugin_module(name,self[name])exceptExceptionase:self.__crashes[name]=PluginCrashed(e)logging.error(f'[{name}]: plugin import failed: {e}')logging.error(traceback.format_exc())self.load_order.remove(name)fornameinlist(self.load_order):try:self.__init_plugin_module(name,self[name])exceptExceptionase:self.__crashes[name]=PluginCrashed(e)logging.error(f'[{name}]: plugin init failed: {e}')logging.error(traceback.format_exc())self.load_order.remove(name)logging.info(f'Loaded {len(self.load_order):d} plugins')