forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlpm.py
More file actions
89 lines (65 loc) · 2.37 KB
/
lpm.py
File metadata and controls
89 lines (65 loc) · 2.37 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
from multiprocessing.pool import Pool
from typing import Callable, Dict
import click
from click import ClickException
from rich.console import Console
from localstack.services.install import InstallerManager
from localstack.utils.bootstrap import setup_logging
console = Console()
@click.group()
def cli():
"""
The LocalStack Package Manager (lpm) CLI is a set of commands to install third-party packages used by localstack
service providers.
Here are some handy commands:
List all packages
python -m localstack.cli.lpm list
Install DynamoDB Local:
python -m localstack.cli.install dynamodb-local
Install all community packages, four in parallel:
python -m localstack.cli.lpm list | grep "/community" | cut -d'/' -f1 | xargs python -m localstack.cli.lpm install --parallel 4
"""
setup_logging()
def _do_install(pkg):
console.print(f"installing... [bold]{pkg}[/bold]")
try:
InstallerManager().get_installers()[pkg]()
console.print(f"[green]installed[/green] [bold]{pkg}[/bold]")
except Exception as e:
console.print(f"[red]error[/red] installing {pkg}: {e}")
raise e
@cli.command()
@click.argument("package", nargs=-1, required=True)
@click.option(
"--parallel",
type=int,
default=1,
required=False,
help="how many installers to run in parallel processes",
)
def install(package, parallel):
"""
Install one or more packages.
"""
console.print(f"resolving packages: {package}")
installers: Dict[str, Callable] = InstallerManager().get_installers()
for pkg in package:
if pkg not in installers:
raise ClickException(f"unable to locate installer for package {pkg}")
if parallel > 1:
console.print(f"install {parallel} packages in parallel:")
# collect installers and install in parallel:
try:
with Pool(processes=parallel) as pool:
pool.map(_do_install, package)
except Exception:
raise ClickException("one or more package installations failed.")
@cli.command(name="list")
def list_packages():
"""List available packages of all repositories"""
installers = InstallerManager()
for repo in installers.repositories.load_all():
for package, _ in repo.get_installer():
console.print(f"[green]{package}[/green]/{repo.name}")
if __name__ == "__main__":
cli()