Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, namespace, driver_list, **kwargs):
super(DriverManager, self).__init__()
manager = stevedore.named.NamedExtensionManager(
namespace, driver_list, invoke_on_load=True, **kwargs)
drivers = {}
for ext in manager:
type_ = ext.obj.get_type()
if type_ in drivers:
msg = _("driver '%(new_driver)s' ignored because "
"driver '%(old_driver)s' is already "
"registered for driver '%(type)s'") % {
'new_driver': ext.name,
'old_driver': drivers[type].name,
'type': type_}
LOG.error(msg)
raise SystemExit(msg)
drivers[type_] = ext
self._drivers = dict((type_, ext.obj)
if parsed_args.modules:
for module_name in parsed_args.modules:
module = _import_module(module_name)
if module:
for group, opts in _list_opts(module):
opts_by_group.setdefault(group, []).append((module_name,
opts))
# Look for entry points defined in libraries (or applications) for
# option discovery, and include their return values in the output.
#
# Each entry point should be a function returning an iterable
# of pairs with the group name (or None for the default group)
# and the list of Opt instances for that group.
if parsed_args.libraries:
loader = stevedore.named.NamedExtensionManager(
'oslo.config.opts',
names=list(set(parsed_args.libraries)),
invoke_on_load=False,
on_load_failure_callback=raise_extension_exception
)
for ext in loader:
for group, opts in ext.plugin():
opt_list = opts_by_group.setdefault(group or 'DEFAULT', [])
opt_list.append((ext.name, opts))
for pkg_name in pkg_names:
mods = mods_by_pkg.get(pkg_name)
mods.sort()
for mod_str in mods:
if mod_str.endswith('.__init__'):
mod_str = mod_str[:mod_str.rfind(".")]
default=DEFAULT_PLUGINS,
help=u._('List of crypto plugins to load.')
)
]
CONF.register_group(crypto_opt_group)
CONF.register_opts(crypto_opts, group=crypto_opt_group)
config.parse_args(CONF)
config.set_module_config("crypto", CONF)
def list_opts():
yield crypto_opt_group, crypto_opts
class _CryptoPluginManager(named.NamedExtensionManager):
def __init__(self, conf=CONF, invoke_args=(), invoke_kwargs={}):
"""Crypto Plugin Manager
Each time this class is initialized it will load a new instance
of each enabled crypto plugin. This is undesirable, so rather than
initializing a new instance of this class use the PLUGIN_MANAGER
at the module level.
"""
crypto_conf = config.get_module_config('crypto')
plugin_names = self._get_internal_plugin_names(crypto_conf)
super(_CryptoPluginManager, self).__init__(
crypto_conf.crypto.namespace,
plugin_names,
invoke_on_load=False, # Defer creating plugins to utility below.
invoke_args=invoke_args,
if parsed_args.modules:
for module_name in parsed_args.modules:
module = _import_module(module_name)
if module:
for group, opts in _list_opts(module):
opts_by_group.setdefault(group, []).append((module_name,
opts))
# Look for entry points defined in libraries (or applications) for
# option discovery, and include their return values in the output.
#
# Each entry point should be a function returning an iterable
# of pairs with the group name (or None for the default group)
# and the list of Opt instances for that group.
if parsed_args.libraries:
loader = stevedore.named.NamedExtensionManager(
'oslo_config.opts',
names=list(set(parsed_args.libraries)),
invoke_on_load=False,
on_load_failure_callback=raise_extension_exception
)
for ext in loader:
for group, opts in ext.plugin():
opt_list = opts_by_group.setdefault(group or 'DEFAULT', [])
opt_list.append((ext.name, opts))
for pkg_name in pkg_names:
mods = mods_by_pkg.get(pkg_name)
mods.sort()
for mod_str in mods:
if mod_str.endswith('.__init__'):
mod_str = mod_str[:mod_str.rfind(".")]
if parsed_args.modules:
for module_name in parsed_args.modules:
module = _import_module(module_name)
if module:
for group, opts in _list_opts(module):
opts_by_group.setdefault(group, []).append((module_name,
opts))
# Look for entry points defined in libraries (or applications) for
# option discovery, and include their return values in the output.
#
# Each entry point should be a function returning an iterable
# of pairs with the group name (or None for the default group)
# and the list of Opt instances for that group.
if parsed_args.libraries:
loader = stevedore.named.NamedExtensionManager(
'oslo.config.opts',
names=list(set(parsed_args.libraries)),
invoke_on_load=False,
on_load_failure_callback=raise_extension_exception
)
for ext in loader:
for group, opts in ext.plugin():
opt_list = opts_by_group.setdefault(group or 'DEFAULT', [])
opt_list.append((ext.name, opts))
for pkg_name in pkg_names:
mods = mods_by_pkg.get(pkg_name)
mods.sort()
for mod_str in mods:
if mod_str.endswith('.__init__'):
mod_str = mod_str[:mod_str.rfind(".")]
def _get_enforcer(namespace):
"""Find a policy.Enforcer via an entry point with the given namespace.
:param namespace: a namespace under oslo.policy.enforcer where the desired
enforcer object can be found.
:returns: a policy.Enforcer object
"""
mgr = stevedore.named.NamedExtensionManager(
'oslo.policy.enforcer',
names=[namespace],
on_load_failure_callback=on_load_failure_callback,
invoke_on_load=True)
if namespace not in mgr:
raise KeyError('Namespace "%s" not found.' % namespace)
enforcer = mgr[namespace].obj
return enforcer
if parsed_args.modules:
for module_name in parsed_args.modules:
module = _import_module(module_name)
if module:
for group, opts in _list_opts(module):
opts_by_group.setdefault(group, []).append((module_name,
opts))
# Look for entry points defined in libraries (or applications) for
# option discovery, and include their return values in the output.
#
# Each entry point should be a function returning an iterable
# of pairs with the group name (or None for the default group)
# and the list of Opt instances for that group.
if parsed_args.libraries:
loader = stevedore.named.NamedExtensionManager(
'oslo_config.opts',
names=list(set(parsed_args.libraries)),
invoke_on_load=False,
on_load_failure_callback=raise_extension_exception
)
for ext in loader:
for group, opts in ext.plugin():
opt_list = opts_by_group.setdefault(group or 'DEFAULT', [])
opt_list.append((ext.name, opts))
for pkg_name in pkg_names:
mods = mods_by_pkg.get(pkg_name)
mods.sort()
for mod_str in mods:
if mod_str.endswith('.__init__'):
mod_str = mod_str[:mod_str.rfind(".")]
def load_extension_options(self, module):
# Note that options loaded this way aren't added to _opts_by_module
loader = stevedore.named.NamedExtensionManager(
'oslo.config.opts',
names=(module,),
invoke_on_load=False
)
for ext in loader:
for group, opts in ext.plugin():
for opt in opts:
if group is None:
self._add_opt(opt.dest, 'DEFAULT', opt)
else:
self._add_opt(group + '/' + opt.dest, group, opt)
self._opt_names.sort(OptionsCache._cmpopts)
if parsed_args.modules:
for module_name in parsed_args.modules:
module = _import_module(module_name)
if module:
for group, opts in _list_opts(module):
opts_by_group.setdefault(group, []).append((module_name,
opts))
# Look for entry points defined in libraries (or applications) for
# option discovery, and include their return values in the output.
#
# Each entry point should be a function returning an iterable
# of pairs with the group name (or None for the default group)
# and the list of Opt instances for that group.
if parsed_args.libraries:
loader = stevedore.named.NamedExtensionManager(
'oslo_config.opts',
names=list(set(parsed_args.libraries)),
invoke_on_load=False,
on_load_failure_callback=raise_extension_exception
)
for ext in loader:
for group, opts in ext.plugin():
opt_list = opts_by_group.setdefault(group or 'DEFAULT', [])
opt_list.append((ext.name, opts))
for pkg_name in pkg_names:
mods = mods_by_pkg.get(pkg_name)
mods.sort()
for mod_str in mods:
if mod_str.endswith('.__init__'):
mod_str = mod_str[:mod_str.rfind(".")]
app.register_error_handler(code, _json_error)
# TODO(kiall): Ideally, we want to make use of the Plugin class here.
# This works for the moment though.
def _register_blueprint(ext):
app.register_blueprint(ext.plugin)
# Add all in-built APIs
mgr = extension.ExtensionManager('designate.api.v1')
mgr.map(_register_blueprint)
# Add any (enabled) optional extensions
extensions = cfg.CONF['service:api'].enabled_extensions_v1
if len(extensions) > 0:
extmgr = named.NamedExtensionManager('designate.api.v1.extensions',
names=extensions)
extmgr.map(_register_blueprint)
return app