Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def processing_hooks_manager(*args):
"""Create a Stevedore extension manager for processing hooks.
:param args: arguments to pass to the hooks constructor.
"""
global _HOOKS_MGR
if _HOOKS_MGR is None:
names = [x.strip()
for x in CONF.processing.processing_hooks.split(',')
if x.strip()]
_HOOKS_MGR = stevedore.NamedExtensionManager(
'ironic_inspector.hooks.processing',
names=names,
invoke_on_load=True,
invoke_args=args,
on_missing_entrypoints_callback=missing_entrypoints_callback,
name_order=True)
return _HOOKS_MGR
The ResourceHandler uses stevedore to load the resource plugins.
The handler can handle and report exceptions raised in the plugins
depending on the value of the propagate_map_exceptions parameter.
It is useful in testing to propagate exceptions so they are exposed
as part of the test. If exceptions are not propagated they are
logged at error level.
Any named plugins that are not located are logged.
:param names: the list of plugins to load by name
:param propagate_map_exceptions: True indicates exceptions in the
plugins should be raised, False indicates they should be handled and
logged.
"""
self._mgr = stevedore.NamedExtensionManager(
namespace=RESOURCE_NAMESPACE,
names=names,
propagate_map_exceptions=propagate_map_exceptions,
invoke_on_load=True)
if self._mgr.names():
LOG.warning(_LW(
'The Extensible Resource Tracker is deprecated and will '
'be removed in the 14.0.0 release. If you '
'use this functionality and have custom resources that '
'are managed by the Extensible Resource Tracker, please '
'contact the Nova development team by posting to the '
'openstack-dev mailing list. There is no future planned '
'support for the tracking of custom resources.'))
self._log_missing_plugins(names)
def __init__(self, *args, **kwargs):
super(Healthcheck, self).__init__(*args, **kwargs)
self.oslo_conf.register_opts(opts.HEALTHCHECK_OPTS,
group='healthcheck')
self._path = self._conf_get('path')
self._show_details = self._conf_get('detailed')
self._backends = stevedore.NamedExtensionManager(
self.NAMESPACE, self._conf_get('backends'),
name_order=True, invoke_on_load=True,
invoke_args=(self.oslo_conf, self.conf))
self._accept_to_functor = collections.OrderedDict([
# Order here matters...
('text/plain', self._make_text_response),
('text/html', self._make_html_response),
('application/json', self._make_json_response),
])
self._accept_order = tuple(self._accept_to_functor)
# When no accept type matches instead of returning 406 we will
# always return text/plain (because sending an error from this
# middleware actually can cause issues).
self._default_accept = 'text/plain'
self._ignore_path = False
make_context.add_argument('-u', '--update', action='store_false',
help='Update the previous docker-context, '
'the files added in the docker-context '
'after the previously pkr make will not '
'be removed')
make_context.set_defaults(func=lambda a: Kard.load_current().make(
reset=a.update))
create_kard_p = sub_p.add_parser('create', help='Create a new kard')
create_kard_p.set_defaults(func=_create_kard)
create_kard_p.add_argument('name', help='The name of the kard')
create_kard_p.add_argument('-e', '--env',
default='dev',
help='The environment (dev/prod)')
entry_points = stevedore.NamedExtensionManager(
'drivers', []).list_entry_points()
create_kard_p.add_argument('-d', '--driver',
default='compose',
help='The pkr driver to use {}'.format(
tuple(entry_point.name
for entry_point in entry_points)))
create_kard_p.add_argument('-m', '--meta',
type=argparse.FileType('r'),
help='A file to load meta from')
create_kard_p.add_argument(
'--features',
help='Comma-separated list of features to include in the deployment. '
'Take one or more of: elk, protractor, salt',
default=None)
def extension_manager(names):
return stevedore.NamedExtensionManager(
_COLLECTOR_NS, names=names, name_order=True,
on_missing_entrypoints_callback=_extension_manager_err_callback)
app.identity_policy_mgr.exempt(views.fetch)
app.identity_policy_mgr.exempt(views.view)
# Default routes. Other routes must be added through App extensions.
# Default routes are set up before app extensions are loaded so extensions
# can introspect/modify view functions.
app.route('/threads//comments', methods=['GET'])(views.fetch)
app.route('/threads//comments', methods=['POST'])(views.new)
app.route('/comments/', methods=['GET'])(views.view)
app.route('/comments/', methods=['PATCH'])(views.edit)
app.route('/comments/', methods=['DELETE'])(views.delete)
app.route('/login', methods=['GET', 'POST'])(views.admin_login)
app.route('/logout', methods=['GET'])(views.admin_logout)
# Load all extensions explicitly enabled via `ENABLE_EXT_*` parameters.
app.ext_mgr = stevedore.NamedExtensionManager(
namespace='pg_discuss.ext',
names=config.get_enabled_extensions(app.config),
name_order=True,
invoke_on_load=True,
invoke_kwds={'app': app},
on_load_failure_callback=ext.fail_on_ext_load,
propagate_map_exceptions=True,
)
# Create hook map
app.hook_map = ext.get_hook_map(app.ext_mgr.extensions, ext.hook_classes())
# Run the `init_app` hooks.
ext.exec_init_app(app)
# Add a route to the landing page at the root, '/'. Ignore if an extension
# has already set up a route for the root.
def _get_app_class(self, app_type):
"""Get an application class (Python class) by app name"""
mgr = stevedore.NamedExtensionManager(
'dragonflow.controller.apps',
[app_type],
invoke_on_load=False,
)
for ext in mgr:
return ext.plugin
else:
raise RuntimeError(_('Failed to load app {0}').format(app_type))