How to use the stevedore.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 openstack / tempest / tempest / test_discover / plugins.py View on Github external
def __init__(self):
        self.ext_plugins = stevedore.ExtensionManager(
            'tempest.test_plugins', invoke_on_load=True,
            propagate_map_exceptions=True,
            on_load_failure_callback=self.failure_hook)
        self._register_service_clients()
github openstack / oslo.config / oslo_config / cfg.py View on Github external
def _open_source_from_opt_group(self, group_name):
        if not self._ext_mgr:
            self._ext_mgr = stevedore.ExtensionManager(
                "oslo.config.driver",
                invoke_on_load=True)

        self.register_opt(
            StrOpt('driver',
                   choices=self._ext_mgr.names(),
                   help=_SOURCE_DRIVER_OPTION_HELP),
            group=group_name)

        try:
            driver_name = self[group_name].driver
        except ConfigFileValueError as err:
            LOG.error(
                "could not load configuration from %r. %s",
                group_name, err.msg)
            return None
github nttcom / eclcli / eclcli / api / auth.py View on Github external
def get_plugin_list():
    """Gather plugin list and cache it"""

    global PLUGIN_LIST

    if PLUGIN_LIST is None:
        PLUGIN_LIST = stevedore.ExtensionManager(
            base.PLUGIN_NAMESPACE,
            invoke_on_load=False,
            propagate_map_exceptions=True,
        )
    return PLUGIN_LIST
github openstack / watcher / watcher / decision_engine / framework / strategy / strategy_loader.py View on Github external
def load_strategies(self):
        extension_manager = ExtensionManager(
            namespace='watcher_strategies',
            invoke_on_load=True,
        )
        return {ext.name: ext.plugin for ext in extension_manager.extensions}
github openstack / ironic-inspector / ironic_inspector / plugins / base.py View on Github external
def rule_conditions_manager():
    """Create a Stevedore extension manager for conditions in rules."""
    global _CONDITIONS_MGR
    if _CONDITIONS_MGR is None:
        _CONDITIONS_MGR = stevedore.ExtensionManager(
            'ironic_inspector.rules.conditions',
            invoke_on_load=True)
    return _CONDITIONS_MGR
github openstack / dragonflow / dragonflow / db / models / all.py View on Github external
def load_all_extensions():
    """Load all available modules with DF models"""
    manager = stevedore.ExtensionManager(  # noqa F841
        'dragonflow.db.models',
    )
github openstack / ironic-inspector / ironic_inspector / plugins / base.py View on Github external
def introspection_data_manager():
    global _INTROSPECTION_DATA_MGR
    if _INTROSPECTION_DATA_MGR is None:
        _INTROSPECTION_DATA_MGR = stevedore.ExtensionManager(
            'ironic_inspector.introspection_data.store',
            invoke_on_load=True)
    return _INTROSPECTION_DATA_MGR
github sqlalchemy / dogpile.cache / dogpile / util / langhelpers.py View on Github external
def load(self, name):
        if name in self._unloaded:
            self.impls[name] = self._unloaded[name]()
            return self.impls[name]
        if name in self.impls:
            return self.impls[name]
        else:  # pragma NO COVERAGE
            if self._mgr is None:
                self._mgr = stevedore.ExtensionManager(self.group)
            try:
                self.impls[name] = self._mgr[name].plugin
                return self.impls[name]
            except KeyError:
                raise self.NotFound(
                    "Can't load plugin %s %s" % (self.group, name)
                )
github python-medikit / medikit / medikit / config / registry.py View on Github external
def __init__(self, dispatcher: EventDispatcher):
        self._configs = {}
        self._features = {}
        self._pipelines = {}
        self._variables = OrderedDict()

        self.dispatcher = dispatcher
        self.resources = OrderedDict()

        def register_feature(ext):
            self._features[ext.name] = ext.plugin

        def on_load_feature_failure(mgr, entrypoint, err):
            logger.exception("Exception caught while loading {}.".format(entrypoint), err)

        mgr = ExtensionManager(namespace="medikit.feature", on_load_failure_callback=on_load_feature_failure)
        mgr.map(register_feature)

        dispatcher.add_listener(medikit.on_end, self.write_resources)
github AdaCore / e3-core / src / e3 / anod / sandbox / main.py View on Github external
'e3.anod.sandbox.sandbox_action': [
                'foo = e3_contrib.sandbox_actions.SandBoxFoo']
        }

    :param get_argument_parser: return e3.main.Main argument_parser instead
        of running the action.
    """
    m = Main()
    m.parse_args(known_args_only=True)

    subparsers = m.argument_parser.add_subparsers(
        title="action", description="valid actions"
    )

    # Load all sandbox actions plugins
    ext = stevedore.ExtensionManager(
        namespace="e3.anod.sandbox.sandbox_action",
        invoke_on_load=True,
        invoke_args=(subparsers,),
    )

    if len(ext.names()) != len(ext.entry_points_names()):
        raise SandBoxError(
            "an error occured when loading sandbox_action entry points %s"
            % ",".join(ext.entry_points_names())
        )  # defensive code

    if get_argument_parser:
        return m.argument_parser

    args = m.argument_parser.parse_args()