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_all_extensions(self):
self.api_extension_manager = stevedore.enabled.EnabledExtensionManager(
namespace=self.EXT_NAMESPACE,
check_func=self._check_load_extension,
invoke_on_load=True,
invoke_kwds={})
self.api_extension_manager.map(self.add_extension)
LOG.debug("Did not find any server create extensions")
# Look for implementation of extension point of server rebuild
self.rebuild_extension_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_REBUILD_NAMESPACE,
check_func=_check_load_extension('server_rebuild'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
if not list(self.rebuild_extension_manager):
LOG.debug("Did not find any server rebuild extensions")
# Look for implementation of extension point of server update
self.update_extension_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_UPDATE_NAMESPACE,
check_func=_check_load_extension('server_update'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
if not list(self.update_extension_manager):
LOG.debug("Did not find any server update extensions")
# Look for implementation of extension point of server resize
self.resize_extension_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_RESIZE_NAMESPACE,
check_func=_check_load_extension('server_resize'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
LOG.debug(
'extension %(ext_alias)s is missing %(func)s',
{'ext_alias': ext.obj.alias,
'func': required_function})
return False
else:
return False
return check_load_extension
self.extension_info = kwargs.pop('extension_info')
super(ServersController, self).__init__(**kwargs)
self.compute_api = compute.API(skip_policy_check=True)
# Look for implementation of extension point of server creation
self.create_extension_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_CREATE_NAMESPACE,
check_func=_check_load_extension('server_create'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
if not list(self.create_extension_manager):
LOG.debug("Did not find any server create extensions")
# Look for implementation of extension point of server rebuild
self.rebuild_extension_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_REBUILD_NAMESPACE,
check_func=_check_load_extension('server_rebuild'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
self.update_schema_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_UPDATE_NAMESPACE,
check_func=_check_load_extension('get_server_update_schema'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
if list(self.update_schema_manager):
self.update_schema_manager.map(self._update_extension_schema,
self.schema_server_update)
else:
LOG.debug("Did not find any server update schemas")
# Look for API schema of server rebuild extension
self.rebuild_schema_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_REBUILD_NAMESPACE,
check_func=_check_load_extension('get_server_rebuild_schema'),
invoke_on_load=True,
invoke_kwds={"extension_info": self.extension_info},
propagate_map_exceptions=True)
if list(self.rebuild_schema_manager):
self.rebuild_schema_manager.map(self._rebuild_extension_schema,
self.schema_server_rebuild)
else:
LOG.debug("Did not find any server rebuild schemas")
# Look for API schema of server resize extension
self.resize_schema_manager = \
stevedore.enabled.EnabledExtensionManager(
namespace=self.EXTENSION_RESIZE_NAMESPACE,
check_func=_check_load_extension('get_server_resize_schema'),
def make_app():
"""App builder (wsgi).
Entry point for Blazar REST API server.
"""
app = flask.Flask('blazar.api')
app.route('/', methods=['GET'])(version_list)
app.route('/versions', methods=['GET'])(version_list)
LOG.debug("List of plugins: %s", cfg.CONF.manager.plugins)
plugins = cfg.CONF.manager.plugins + ['leases']
extension_manager = enabled.EnabledExtensionManager(
check_func=lambda ext: ext.name in plugins,
namespace='blazar.api.v1.extensions',
invoke_on_load=False
)
for ext in extension_manager.extensions:
bp = ext.plugin()
app.register_blueprint(bp, url_prefix=bp.url_prefix)
for code in werkzeug_exceptions.default_exceptions:
app.register_error_handler(code, make_json_error)
if cfg.CONF.debug and not cfg.CONF.log_exchange:
LOG.debug('Logging of request/response exchange could be enabled '
'using flag --log_exchange')