-
Notifications
You must be signed in to change notification settings - Fork 0
/
load_as_extension.py
46 lines (29 loc) · 1012 Bytes
/
load_as_extension.py
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
from stevedore import extension
mgr = extension.ExtensionManager(
namespace='stevedore_testplugins.database',
invoke_on_load=True,
invoke_args=(),
)
def func1(ext, data):
return (ext.name, ext.obj.format(data))
results = mgr.map(func1, "")
for ext , result in results:
print(f"return value from extension ({ext}) is ({result})")
'''
from pkg_resources import iter_entry_points
_plugins = {}
def load_register_plugins(group='general'):
group=f"stevedore_testplugins.{group}"
if group not in _plugins:
_plugins[group] = []
end_points = iter_entry_points(group=group)
for end_point in end_points:
if end_point.name in _plugins[group]:
raise ValueError(f"The plugin {end_point.name} already registered!")
_plugins[group].append(end_point)
print(f"\nEntry points of group {group} are:")
for plugin in _plugins[group]:
function = plugin.load()
function()
load_register_plugins(group = "database")
'''