forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathplugins.py
More file actions
134 lines (102 loc) Β· 3.42 KB
/
plugins.py
File metadata and controls
134 lines (102 loc) Β· 3.42 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
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
import os
import time
import click
from plux import PluginManager
from plux.build.setuptools import find_plugins
from plux.core.entrypoint import spec_to_entry_point
from rich import print as rprint
from rich.console import Console
from rich.table import Table
from rich.tree import Tree
from localstack.cli.exceptions import CLIError
console = Console()
@click.group()
def cli():
"""
The plugins CLI is a set of commands to help troubleshoot LocalStack's plugin mechanism.
"""
pass
@cli.command()
@click.option("--where", type=str, default=os.path.abspath(os.curdir))
@click.option("--exclude", multiple=True, default=())
@click.option("--include", multiple=True, default=("*",))
@click.option("--output", type=str, default="tree")
def find(where, exclude, include, output):
"""
Find plugins by scanning the given path for PluginSpecs.
It starts from the current directory if --where is not specified.
This is what a setup.py method would run as a build step, i.e., discovering entry points.
"""
with console.status(f"Scanning path {where}"):
plugins = find_plugins(where, exclude, include)
if output == "tree":
tree = Tree("Entrypoints")
for namespace, entry_points in plugins.items():
node = tree.add(f"[bold]{namespace}")
t = Table()
t.add_column("Name")
t.add_column("Location")
for ep in entry_points:
key, value = ep.split("=")
t.add_row(key, value)
node.add(t)
rprint(tree)
elif output == "dict":
rprint(dict(plugins))
else:
raise CLIError(f"unknown output format {output}")
@cli.command("list")
@click.option("--namespace", type=str, required=True)
def cmd_list(namespace):
"""
List all available plugins using a PluginManager from available endpoints.
"""
manager = PluginManager(namespace)
t = Table()
t.add_column("Name")
t.add_column("Factory")
for spec in manager.list_plugin_specs():
ep = spec_to_entry_point(spec)
t.add_row(spec.name, ep.value)
rprint(t)
@cli.command()
@click.option("--namespace", type=str, required=True)
@click.option("--name", type=str, required=True)
def load(namespace, name):
"""
Attempts to load a plugin using a PluginManager.
"""
manager = PluginManager(namespace)
with console.status(f"Loading {namespace}:{name}"):
then = time.time()
plugin = manager.load(name)
took = time.time() - then
rprint(
f":tada: successfully loaded [bold][green]{namespace}[/green][/bold]:[bold][cyan]{name}[/cyan][/bold] ({type(plugin)}"
)
rprint(f":stopwatch: loading took {took:.4f} s")
@cli.command()
@click.option("--namespace", type=str)
def cache(namespace):
"""
Outputs the stevedore entrypoints cache from which plugins are loaded.
"""
from stevedore._cache import _c
data = _c._get_data_for_path(None)
tree = Tree("Entrypoints")
for group, entry_points in data.get("groups").items():
if namespace and group != namespace:
continue
node = tree.add(f"[bold]{group}")
t = Table()
t.add_column("Name")
t.add_column("Value")
for key, value, _ in entry_points:
t.add_row(key, value)
node.add(t)
if namespace:
rprint(t)
return
rprint(tree)
if __name__ == "__main__":
cli()