How to use the stevedore.extension.ExtensionManager function in stevedore

To help you get started, we’ve selected a few stevedore examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github unitedstack / gringotts / gringotts / middleware / nova.py View on Github external
self.server_action_regex,
            self.attach_volume_to_server_regex,
        ]
        self.resize_resource_actions = [
            self.resize_server_action,
        ]
        self.stop_resource_actions = [
            self.stop_server_action,
        ]
        self.start_resource_actions = [
            self.start_server_action,
        ]
        self.restore_resource_actions = [
            self.restore_server_action,
        ]
        self.product_items = extension.ExtensionManager(
            namespace='gringotts.server.product_items',
            invoke_on_load=True,
            invoke_args=(self.gclient,))
github openstack / mistral / mistral / services / action_manager.py View on Github external
def _register_dynamic_action_classes(namespace=''):
    extensions = extension.ExtensionManager(
        namespace='mistral.generators',
        invoke_on_load=True
    )

    for ext in extensions:
        for generator in ext.obj:
            _register_actions(generator, namespace)
github eonpatapon / contrail-api-cli / contrail_api_cli / manager.py View on Github external
def load_namespace(self, ns):
        """Load commands from namespace.

        :param ns: namespace name
        :type ns: str
        """
        mgr = extension.ExtensionManager(namespace=ns,
                                         verify_requirements=True,
                                         on_load_failure_callback=self._on_failure)
        # Load commands from ns
        for ext in mgr.extensions:
            try:
                obj = ext.plugin(ext.name)
            except Exception as err:
                self._on_failure(self, ext, err)
                obj = None
            ext.obj = obj

        self.mgrs.append(mgr)
github StackStorm / orquesta / orquesta / expressions / base.py View on Github external
def get_evaluators():
    global _EXP_EVALUATORS

    if _EXP_EVALUATORS is None:
        _EXP_EVALUATORS = {}

        mgr = extension.ExtensionManager(
            namespace=_EXP_EVALUATOR_NAMESPACE,
            invoke_on_load=False
        )

        for name in mgr.names():
            _EXP_EVALUATORS[name] = get_evaluator(name)

    return _EXP_EVALUATORS
github openstack / cloudkitty / cloudkitty / transformer / __init__.py View on Github external
def get_transformers():
    transformers = {}
    transformer_exts = extension.ExtensionManager(
        TRANSFORMERS_NAMESPACE,
        invoke_on_load=True)
    for transformer in transformer_exts:
        t_name = transformer.name
        t_obj = transformer.obj
        transformers[t_name] = t_obj
    return transformers
github StackStorm / st2 / st2auth / st2auth / backends / __init__.py View on Github external
def get_available_backends():
    """
    Return names of the available / installed authentication backends.

    :rtype: ``list`` of ``str``
    """
    from stevedore.extension import ExtensionManager

    manager = ExtensionManager(namespace=BACKENDS_NAMESPACE, invoke_on_load=False)
    return manager.names()
github PyCQA / bandit / bandit / core / extension_loader.py View on Github external
def load_plugins(self, plugins_namespace):
        self.plugins_mgr = extension.ExtensionManager(
            namespace=plugins_namespace,
            invoke_on_load=False,
            verify_requirements=False,
            )

        def test_has_id(plugin):
            if not hasattr(plugin.plugin, "_test_id"):
                # logger not setup yet, so using print
                print("WARNING: Test '%s' has no ID, skipping." % plugin.name,
                      file=sys.stderr)
                return False
            return True

        self.plugins = list(filter(test_has_id, list(self.plugins_mgr)))
        self.plugin_names = [plugin.name for plugin in self.plugins]
        self.plugins_by_id = {p.plugin._test_id: p for p in self.plugins}
github openstack / python-ironicclient / ironicclient / openstack / common / apiclient / auth.py View on Github external
def discover_auth_systems():
    """Discover the available auth-systems.

    This won't take into account the old style auth-systems.
    """
    global _discovered_plugins
    _discovered_plugins = {}

    def add_plugin(ext):
        _discovered_plugins[ext.name] = ext.plugin

    ep_namespace = "ironicclient.openstack.common.apiclient.auth"
    mgr = extension.ExtensionManager(ep_namespace)
    mgr.map(add_plugin)
github openstack / python-magnumclient / magnumclient / openstack / common / apiclient / auth.py View on Github external
def discover_auth_systems():
    """Discover the available auth-systems.

    This won't take into account the old style auth-systems.
    """
    global _discovered_plugins
    _discovered_plugins = {}

    def add_plugin(ext):
        _discovered_plugins[ext.name] = ext.plugin

    ep_namespace = "magnumclient.openstack.common.apiclient.auth"
    mgr = extension.ExtensionManager(ep_namespace)
    mgr.map(add_plugin)
github openstack / monasca-transform / monasca_transform / transform / builder / generic_transform_builder.py View on Github external
def _get_setter_component_manager():
        """stevedore extension manager for setter components."""
        return extension.ExtensionManager(
            namespace=GenericTransformBuilder.
            _MONASCA_TRANSFORM_SETTER_NAMESPACE,
            on_load_failure_callback=GenericTransformBuilder.
            log_load_extension_error,
            invoke_on_load=False)