Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def load_plugins(self):
session = db.Session()
deactivate_all_plugins(session)
session.commit()
default_paths = self.get_all_plugin_paths()
# Load plugins from directories
plugin_base = PluginBase(package="deeptracy.plugins")
plugin_manager = plugin_base.make_plugin_source(searchpath=default_paths)
# Locate plugins
plugins_found = {}
for module in plugin_manager.list_plugins():
if module == 'store':
continue
module_objects = plugin_manager.load_plugin(module)
for plugin_name, plugin_obj in vars(module_objects).items():
if plugin_name.startswith("_") or type(plugin_obj).__name__ != "function":
continue
def load(self):
self.discord_commands = dict()
self.discord_callbacks = dict()
self.plugins = dict()
# Resets the dicts.
self.plugin_base = PluginBase(package='__main__.plugins')
# Defines the plugin base for Discord functions.
self.plugin_source = self.plugin_base.make_plugin_source(
searchpath=['./plugins'])
# Defines the plugin source for Discord functions.
listening_events = ["on_message", "on_ready", "on_message_delete", "on_reaction_add", "on_reaction_remove", "on_channel_delete", "on_channel_create",
"on_channel_update", "on_member_join", "on_member_remove", "on_member_update", "on_server_join", "on_server_remove", "on_server_update"]
# The list of events to look for and seperate.
for plugin in self.plugin_source.list_plugins():
self.plugins[plugin] = self.plugin_source.load_plugin(plugin)
for i in self.plugins[plugin].__dict__.keys():
if callable(self.plugins[plugin].__dict__[i]):
if i == "execute_on_init":
self.plugins[plugin].__dict__[i](self)
import os
from pluginbase import PluginBase
from logbook import Logger
from . import plugins
from .depgraph import DependencyGraph
from .exc import InvocationError
plugin_base = PluginBase(package='unleash.plugins')
log = Logger('plugins')
class PluginGraph(DependencyGraph):
NAME_ATTR = 'PLUGIN_NAME'
DEP_ATTR = 'PLUGIN_DEPENDS'
def __init__(self, *args, **kwargs):
super(PluginGraph, self).__init__(*args, **kwargs)
self.plugin_mods = {}
def add_plugin(self, plugin):
name = getattr(plugin, self.NAME_ATTR)
self.plugin_mods[name] = plugin
self.add_obj(name, depends_on=getattr(plugin, self.DEP_ATTR, []))
def _initialize_plugin_source(self):
plugin_base = PluginBase(package='cis.plugins.validation')
plugin_source = plugin_base.make_plugin_source(
searchpath=[
os.path.join(
os.path.abspath(
os.path.dirname(__file__)
),
'../plugins/validation/'
)
]
)
return plugin_source
import importlib.util
from pluginbase import PluginBase, PluginSource
class MiPluginSource(PluginSource):
def load_plugin(self, name):
plugin = super().load_plugin(name)
spec = importlib.util.spec_from_file_location(self.base.package + '.' + name, plugin.__file__)
plugin = importlib.util.module_from_spec(spec)
spec.loader.exec_module(plugin)
return plugin
class MiPluginBase(PluginBase):
def make_plugin_source(self, *args, **kwargs):
return MiPluginSource(self, *args, **kwargs)
def __init__(self, searchpath=None):
searchpath = searchpath or []
searchpath.append(get_path('plugins'))
self.source = pluginbase.PluginBase(package='protocon.plugins').make_plugin_source(
searchpath=searchpath
)
self.connection_drivers = {}
self.transcoders = {}
for plugin in self.source.list_plugins():
module = self.source.load_plugin(plugin)
if hasattr(module, 'ConnectionDriver'):
self.connection_drivers[plugin] = module.ConnectionDriver
if hasattr(module, 'Transcoder'):
self.transcoders[plugin] = module.Transcoder
def load_plugins():
here = os.path.abspath(os.path.dirname(__file__))
get_path = partial(os.path.join, here)
plugin_dir = get_path('plugins')
plugin_base = PluginBase(
package='wafw00f.plugins', searchpath=[plugin_dir]
)
plugin_source = plugin_base.make_plugin_source(
searchpath=[plugin_dir], persist=True
)
plugin_dict = {}
for plugin_name in plugin_source.list_plugins():
plugin_dict[plugin_name] = plugin_source.load_plugin(plugin_name)
return plugin_dict
def __init__(self, frmwk, searchpath):
self.logger = logging.getLogger('termineter.module_manager')
self.frmwk = frmwk
self.source = pluginbase.PluginBase(package='termineter.modules').make_plugin_source(
searchpath=searchpath
)
self._modules = {}
for module_id in self.source.list_plugins():
self._init_pymodule(self.source.load_plugin(module_id))
def import_plugins(plugin_mount, plugin_base_dir):
plugin_base = PluginBase(package=plugin_mount)
plugin_src_dirs = _get_plugin_src_dirs(plugin_base_dir)
return plugin_base.make_plugin_source(searchpath=plugin_src_dirs)
Args:
extra_policies_path (str): Extra path to find plugins in
Returns:
PluginSource: PluginBase PluginSource for finding plugins
"""
here = os.path.dirname(os.path.realpath(__file__))
default_path = "{}/policies".format(here)
all_paths = [default_path]
if extra_policies_path:
if not os.path.isdir(extra_policies_path):
raise InvalidPoliciesDirectory
all_paths.append(extra_policies_path)
LOG.info("Searching for policies in %s", str(all_paths))
plugin_base = PluginBase(package='lavatory.policy_plugins')
plugin_source = plugin_base.make_plugin_source(searchpath=all_paths)
LOG.debug("Policies found: %s", str(plugin_source.list_plugins()))
return plugin_source