Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_provisioning_manager():
from stevedore import dispatch
config = get_config()
if config.get('PLUGIN_WHITELIST', ''):
plugin_whitelist = config.get('PLUGIN_WHITELIST').split()
mgr = dispatch.NameDispatchExtensionManager(
namespace='pouta_blueprints.drivers.provisioning',
check_func=lambda x: x.name in plugin_whitelist,
invoke_on_load=True,
invoke_args=(logger, get_config()),
)
else:
mgr = dispatch.NameDispatchExtensionManager(
namespace='pouta_blueprints.drivers.provisioning',
check_func=lambda x: True,
invoke_on_load=True,
invoke_args=(logger, get_config()),
)
logger.debug('provisioning manager loaded, extensions: %s ' % mgr.names())
return mgr
"""
from stevedore import dispatch
dynamic_config = get_dynamic_config()
if dynamic_config.get('PLUGIN_WHITELIST'):
plugin_whitelist = dynamic_config.get('PLUGIN_WHITELIST').split()
mgr = dispatch.NameDispatchExtensionManager(
namespace='pebbles.drivers.provisioning',
check_func=lambda x: x.name in plugin_whitelist,
invoke_on_load=True,
invoke_args=(logger, dynamic_config),
)
else:
# ahem, load all plugins if string is empty or not available?
# is this wise? -jyrsa 2016-11-28
mgr = dispatch.NameDispatchExtensionManager(
namespace='pebbles.drivers.provisioning',
check_func=lambda x: True,
invoke_on_load=True,
invoke_args=(logger, dynamic_config),
)
logger.debug('provisioning manager loaded, extensions: %s ' % mgr.names())
return mgr
def get_provisioning_manager():
from stevedore import dispatch
config = get_config()
if config.get('PLUGIN_WHITELIST', ''):
plugin_whitelist = config.get('PLUGIN_WHITELIST').split()
mgr = dispatch.NameDispatchExtensionManager(
namespace='pouta_blueprints.drivers.provisioning',
check_func=lambda x: x.name in plugin_whitelist,
invoke_on_load=True,
invoke_args=(logger, get_config()),
)
else:
mgr = dispatch.NameDispatchExtensionManager(
namespace='pouta_blueprints.drivers.provisioning',
check_func=lambda x: True,
invoke_on_load=True,
invoke_args=(logger, get_config()),
)
logger.debug('provisioning manager loaded, extensions: %s ' % mgr.names())
return mgr
def get_provisioning_manager():
""" Gets a stevedore provisioning manager which controls access to all the
possible drivers.
Manager is used to call a method on one or more of the drivers.
"""
from stevedore import dispatch
dynamic_config = get_dynamic_config()
if dynamic_config.get('PLUGIN_WHITELIST'):
plugin_whitelist = dynamic_config.get('PLUGIN_WHITELIST').split()
mgr = dispatch.NameDispatchExtensionManager(
namespace='pebbles.drivers.provisioning',
check_func=lambda x: x.name in plugin_whitelist,
invoke_on_load=True,
invoke_args=(logger, dynamic_config),
)
else:
# ahem, load all plugins if string is empty or not available?
# is this wise? -jyrsa 2016-11-28
mgr = dispatch.NameDispatchExtensionManager(
namespace='pebbles.drivers.provisioning',
check_func=lambda x: True,
invoke_on_load=True,
invoke_args=(logger, dynamic_config),
)
logger.debug('provisioning manager loaded, extensions: %s ' % mgr.names())
def __init__(self, namespace="io.murano.extensions"):
LOG.info('Loading extension plugins')
self.namespace = namespace
extension_manager = dispatch.EnabledExtensionManager(
self.namespace,
PluginLoader.is_plugin_enabled,
on_load_failure_callback=PluginLoader._on_load_failure)
self.packages = {}
name_map = {}
for ext in extension_manager.extensions:
self.load_extension(ext, name_map)
self.cleanup_duplicates(name_map)
def __init__(self):
LOG.info('Loading package type plugins')
extension_manager = dispatch.EnabledExtensionManager(
NAMESPACE,
self._is_plugin_enabled,
on_load_failure_callback=self._on_load_failure)
self.formats = {}
for ext in extension_manager.extensions:
self._load_plugin(ext)