forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathhooks.py
More file actions
79 lines (61 loc) · 2.39 KB
/
hooks.py
File metadata and controls
79 lines (61 loc) · 2.39 KB
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
import functools
from plugin import PluginManager, plugin
# plugin namespace constants
HOOKS_CONFIGURE_LOCALSTACK_CONTAINER = "localstack.hooks.configure_localstack_container"
HOOKS_INSTALL = "localstack.hooks.install"
HOOKS_ON_INFRA_READY = "localstack.hooks.on_infra_ready"
HOOKS_ON_INFRA_START = "localstack.hooks.on_infra_start"
HOOKS_PREPARE_HOST = "localstack.hooks.prepare_host"
def hook(namespace: str, priority: int = 0, **kwargs):
"""
Decorator for creating functional plugins that have a hook_priority attribute.
"""
def wrapper(fn):
fn.hook_priority = priority
return plugin(namespace=namespace, **kwargs)(fn)
return wrapper
def hook_spec(namespace: str):
"""
Creates a new hook decorator bound to a namespace.
on_infra_start = hook_spec("localstack.hooks.on_infra_start")
@on_infra_start()
def foo():
pass
# run all hooks in order
on_infra_start.run()
"""
fn = functools.partial(hook, namespace=namespace)
# attach hook manager and run method to decorator for convenience calls
fn.manager = HookManager(namespace)
fn.run = fn.manager.run_in_order
return fn
class HookManager(PluginManager):
def load_all_sorted(self, propagate_exceptions=False):
"""
Loads all hook plugins and sorts them by their hook_priority attribute.
"""
plugins = self.load_all(propagate_exceptions)
# the hook_priority attribute is part of the function wrapped in the FunctionPlugin
plugins.sort(
key=lambda _fn_plugin: getattr(_fn_plugin.fn, "hook_priority", 0), reverse=True
)
return plugins
def run_in_order(self, *args, **kwargs):
"""
Loads and runs all plugins in order them with the given arguments.
"""
for fn_plugin in self.load_all_sorted():
fn_plugin(*args, **kwargs)
def __str__(self):
return "HookManager(%s)" % self.namespace
def __repr__(self):
return self.__str__()
# localstack container configuration (on the host)
configure_localstack_container = hook_spec(HOOKS_CONFIGURE_LOCALSTACK_CONTAINER)
# additional installers
install = hook_spec(HOOKS_INSTALL)
# prepare the host that's starting localstack
prepare_host = hook_spec(HOOKS_PREPARE_HOST)
# infra (runtime) lifecycle hooks
on_infra_start = hook_spec(HOOKS_ON_INFRA_START)
on_infra_ready = hook_spec(HOOKS_ON_INFRA_READY)