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):
enabled_ext = cfg.CONF['service:api'].enabled_extensions_admin
if len(enabled_ext) > 0:
self._mgr = named.NamedExtensionManager(
namespace='designate.api.admin.extensions',
names=enabled_ext,
invoke_on_load=True)
for ext in self._mgr:
controller = self
path = ext.obj.get_path()
LOG.info("Registering an API extension at path %s", path)
for p in path.split('.')[:-1]:
if p != '':
controller = getattr(controller, p)
setattr(controller, path.split('.')[-1], ext.obj)
def __init__(self):
enabled_ext = cfg.CONF['service:api'].enabled_extensions_v2
if len(enabled_ext) > 0:
self._mgr = named.NamedExtensionManager(
namespace='designate.api.v2.extensions',
names=enabled_ext,
invoke_on_load=True)
for ext in self._mgr:
controller = self
path = ext.obj.get_path()
for p in path.split('.')[:-1]:
if p != '':
controller = getattr(controller, p)
setattr(controller, path.split('.')[-1], 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(".")]
self.publisher_id = publisher_id
if retry is not None:
self.retry = retry
else:
self.retry = conf.oslo_messaging_notifications.retry
self._driver_names = ([driver] if driver is not None else
conf.oslo_messaging_notifications.driver)
if topics is not None:
self._topics = topics
else:
self._topics = conf.oslo_messaging_notifications.topics
self._serializer = serializer or msg_serializer.NoOpSerializer()
self._driver_mgr = named.NamedExtensionManager(
'oslo.messaging.notify.drivers',
names=self._driver_names,
invoke_on_load=True,
invoke_args=[conf],
invoke_kwds={
'topics': self._topics,
'transport': self.transport,
}
def api_app(conf):
cfg.CONF(args=[], project='monasca')
log_levels = (cfg.CONF.default_log_levels)
cfg.set_defaults(log.log_opts, default_log_levels=log_levels)
log.setup('monasca')
dispatcher_manager = named.NamedExtensionManager(
namespace=DISPATCHER_NAMESPACE,
names=cfg.CONF.dispatcher,
invoke_on_load=True,
invoke_args=[cfg.CONF])
if not list(dispatcher_manager):
LOG.error('Failed to load any dispatchers for %s' %
DISPATCHER_NAMESPACE)
return None
# Create the application
app = resource_api.ResourceAPI()
# add each dispatcher to the application to serve requests offered by
# each dispatcher
for driver in dispatcher_manager:
def _get_opt_default_updaters(namespaces):
mgr = stevedore.named.NamedExtensionManager(
'oslo.config.opts.defaults',
names=namespaces,
warn_on_missing_entrypoint=False,
on_load_failure_callback=on_load_failure_callback,
invoke_on_load=False)
return [ep.plugin for ep in mgr]
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_import_flows(**kwargs):
# NOTE(flaper87): Until we have a better infrastructure to enable
# and disable tasks plugins, hard-code the tasks we know exist,
# instead of loading everything from the namespace. This guarantees
# both, the load order of these plugins and the fact that no random
# plugins will be added/loaded until we feel comfortable with this.
# Future patches will keep using NamedExtensionManager but they'll
# rely on a config option to control this process.
extensions = named.NamedExtensionManager('glance.flows.import',
names=['ovf_process',
'convert',
'introspect'],
name_order=True,
invoke_on_load=True,
invoke_kwds=kwargs)
for ext in extensions.extensions:
yield ext.obj
def _get_driver_opts_loaders(namespaces, driver_option_name):
mgr = stevedore.named.NamedExtensionManager(
namespace='oslo.config.opts.' + driver_option_name,
names=namespaces,
on_load_failure_callback=on_load_failure_callback,
invoke_on_load=False)
return [(e.name, e.plugin) for e in mgr]