forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathplugin.py
More file actions
74 lines (51 loc) · 1.86 KB
/
plugin.py
File metadata and controls
74 lines (51 loc) · 1.86 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
import abc
import logging
import os
import click
LOG = logging.getLogger(__name__)
class Plugin:
# TODO: extract as base class for a localstack plugin
name: str
""" the plugin name which has to be unique within the plugin namespace """
def is_active(self):
return True
def load(self):
"""
Runs plugin loading logic if is_active returns True.
"""
pass
class LocalstackCli:
group: click.Group
def __call__(self, *args, **kwargs):
self.group(*args, **kwargs)
class LocalstackCliPlugin(Plugin):
namespace = "localstack.plugins.cli"
@abc.abstractmethod
def attach(self, cli) -> None:
"""
Attach commands to the `localstack` CLI.
:param cli: the cli object
"""
def load_cli_plugins(cli):
# TODO: generalize into a loading mechanism together with plugin
from stevedore.extension import ExtensionManager
if os.environ.get("DEBUG_PLUGINS", "0").lower() in ("true", "1"):
# importing localstack.config is still quite expensive...
logging.basicConfig(level=logging.DEBUG)
namespace = LocalstackCliPlugin.namespace
# this line actually imports the plugin code
manager = ExtensionManager(namespace=namespace, invoke_on_load=False)
for name, ext in manager.items():
LOG.debug("loading plugin %s:%s type: %s", namespace, name, ext.plugin)
try:
plugin: LocalstackCliPlugin = ext.plugin()
if not plugin.is_active():
LOG.debug("plugin %s is deactivated, skipping")
continue
LOG.info("loading %s:%s", namespace, name)
plugin.load()
LOG.info("attaching to CLI %s:%s", namespace, name)
plugin.attach(cli)
except Exception:
LOG.exception("error loading plugin %s:%s", namespace, name)
pass