/usr/share/pyshared/stevedore/dispatch.py is in python-stevedore 0.14.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | import logging
from .enabled import EnabledExtensionManager
LOG = logging.getLogger(__name__)
class DispatchExtensionManager(EnabledExtensionManager):
"""Loads all plugins and filters on execution.
This is useful for long-running processes that need to pass
different inputs to different extensions.
:param namespace: The namespace for the entry points.
:type namespace: str
:param check_func: Function to determine which extensions to load.
:type check_func: callable
:param invoke_on_load: Boolean controlling whether to invoke the
object returned by the entry point after the driver is loaded.
:type invoke_on_load: bool
:param invoke_args: Positional arguments to pass when invoking
the object returned by the entry point. Only used if invoke_on_load
is True.
:type invoke_args: tuple
:param invoke_kwds: Named arguments to pass when invoking
the object returned by the entry point. Only used if invoke_on_load
is True.
:type invoke_kwds: dict
:param propagate_map_exceptions: Boolean controlling whether exceptions
are propagated up through the map call or whether they are logged and
then ignored
:type invoke_on_load: bool
"""
def map(self, filter_func, func, *args, **kwds):
"""Iterate over the extensions invoking func() for any where
filter_func() returns True.
The signature of filter_func() should be::
def filter_func(ext, *args, **kwds):
pass
The first argument to filter_func(), 'ext', is the
:class:`~stevedore.extension.Extension`
instance. filter_func() should return True if the extension
should be invoked for the input arguments.
The signature for func() should be::
def func(ext, *args, **kwds):
pass
The first argument to func(), 'ext', is the
:class:`~stevedore.extension.Extension` instance.
Exceptions raised from within func() are propagated up and
processing stopped if self.propagate_map_exceptions is True,
otherwise they are logged and ignored.
:param filter_func: Callable to test each extension.
:param func: Callable to invoke for each extension.
:param args: Variable arguments to pass to func()
:param kwds: Keyword arguments to pass to func()
:returns: List of values returned from func()
"""
if not self.extensions:
# FIXME: Use a more specific exception class here.
raise RuntimeError('No %s extensions found' % self.namespace)
response = []
for e in self.extensions:
if filter_func(e, *args, **kwds):
self._invoke_one_plugin(response.append, func, e, args, kwds)
return response
def map_method(self, filter_func, method_name, *args, **kwds):
"""Iterate over the extensions invoking each one's object method called
`method_name` for any where filter_func() returns True.
This is equivalent of using :meth:`map` with func set to
`lambda x: x.obj.method_name()`
while being more convenient.
Exceptions raised from within the called method are propagated up
and processing stopped if self.propagate_map_exceptions is True,
otherwise they are logged and ignored.
.. versionadded:: 0.12
:param filter_func: Callable to test each extension.
:param method_name: The extension method name to call
for each extension.
:param args: Variable arguments to pass to method
:param kwds: Keyword arguments to pass to method
:returns: List of values returned from methods
"""
return self.map(filter_func, self._call_extension_method,
method_name, *args, **kwds)
class NameDispatchExtensionManager(DispatchExtensionManager):
"""Loads all plugins and filters on execution.
This is useful for long-running processes that need to pass
different inputs to different extensions and can predict the name
of the extensions before calling them.
The check_func argument should return a boolean, with ``True``
indicating that the extension should be loaded and made available
and ``False`` indicating that the extension should be ignored.
:param namespace: The namespace for the entry points.
:type namespace: str
:param check_func: Function to determine which extensions to load.
:type check_func: callable
:param invoke_on_load: Boolean controlling whether to invoke the
object returned by the entry point after the driver is loaded.
:type invoke_on_load: bool
:param invoke_args: Positional arguments to pass when invoking
the object returned by the entry point. Only used if invoke_on_load
is True.
:type invoke_args: tuple
:param invoke_kwds: Named arguments to pass when invoking
the object returned by the entry point. Only used if invoke_on_load
is True.
:type invoke_kwds: dict
:param propagate_map_exceptions: Boolean controlling whether exceptions
are propagated up through the map call or whether they are logged and
then ignored
:type invoke_on_load: bool
:param on_load_failure_callback: Callback function that will be called when
a entrypoint can not be loaded. The arguments that will be provided
when this is called (when an entrypoint fails to load) are
(manager, entrypoint, exception)
:type on_load_failure_callback: function
:param verify_requirements: Use setuptools to enforce the
dependencies of the plugin(s) being loaded. Defaults to False.
:type verify_requirements: bool
"""
def __init__(self, namespace, check_func, invoke_on_load=False,
invoke_args=(), invoke_kwds={},
propagate_map_exceptions=False,
on_load_failure_callback=None,
verify_requirements=False):
super(NameDispatchExtensionManager, self).__init__(
namespace=namespace,
check_func=check_func,
invoke_on_load=invoke_on_load,
invoke_args=invoke_args,
invoke_kwds=invoke_kwds,
propagate_map_exceptions=propagate_map_exceptions,
on_load_failure_callback=on_load_failure_callback,
verify_requirements=verify_requirements,
)
def _init_plugins(self, extensions):
super(NameDispatchExtensionManager, self)._init_plugins(extensions)
self.by_name = dict((e.name, e) for e in self.extensions)
def map(self, names, func, *args, **kwds):
"""Iterate over the extensions invoking func() for any where
the name is in the given list of names.
The signature for func() should be::
def func(ext, *args, **kwds):
pass
The first argument to func(), 'ext', is the
:class:`~stevedore.extension.Extension` instance.
Exceptions raised from within func() are propagated up and
processing stopped if self.propagate_map_exceptions is True,
otherwise they are logged and ignored.
:param names: List or set of name(s) of extension(s) to invoke.
:param func: Callable to invoke for each extension.
:param args: Variable arguments to pass to func()
:param kwds: Keyword arguments to pass to func()
:returns: List of values returned from func()
"""
response = []
for name in names:
try:
e = self.by_name[name]
except KeyError:
LOG.debug('Missing extension %r being ignored', name)
else:
self._invoke_one_plugin(response.append, func, e, args, kwds)
return response
def map_method(self, names, method_name, *args, **kwds):
"""Iterate over the extensions invoking each one's object method called
`method_name` for any where the name is in the given list of names.
This is equivalent of using :meth:`map` with func set to
`lambda x: x.obj.method_name()`
while being more convenient.
Exceptions raised from within the called method are propagated up
and processing stopped if self.propagate_map_exceptions is True,
otherwise they are logged and ignored.
.. versionadded:: 0.12
:param names: List or set of name(s) of extension(s) to invoke.
:param method_name: The extension method name
to call for each extension.
:param args: Variable arguments to pass to method
:param kwds: Keyword arguments to pass to method
:returns: List of values returned from methods
"""
return self.map(names, self._call_extension_method,
method_name, *args, **kwds)
|