Repository navigation
Expand file tree
/
Copy pathcli.py
More file actions
92 lines (75 loc) · 2.51 KB
/
Copy pathcli.py
File metadata and controls
92 lines (75 loc) · 2.51 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
from __future__ import annotations
import sys
import click
from loguru import logger
from .. import cli_help as ch
from .manager import StackError, StackManager
@click.group(
help=ch.CMD_DAEMON_GROUP,
short_help=ch.CMD_DAEMON_GROUP,
epilog=ch.EPILOG_DAEMON,
no_args_is_help=True,
)
def cli() -> None:
pass
def _print_status(mgr: StackManager) -> None:
status = mgr.status()
click.echo(f"state: {status.state.value}")
click.echo(
f"memgraph: {status.memgraph_endpoint} (reachable={status.memgraph_reachable})"
)
click.echo(
f"qdrant: {status.qdrant_endpoint} (reachable={status.qdrant_reachable})"
)
click.echo(f"compose: {status.compose_file}")
@cli.command("up", help=ch.CMD_DAEMON_UP, short_help=ch.CMD_DAEMON_UP)
def up_cmd() -> None:
mgr = StackManager()
try:
mgr.ensure_running()
_print_status(mgr)
except StackError as e:
logger.error(str(e))
click.secho(str(e), fg="red", err=True)
sys.exit(1)
@cli.command("down", help=ch.CMD_DAEMON_DOWN, short_help=ch.CMD_DAEMON_DOWN)
def down_cmd() -> None:
mgr = StackManager()
try:
mgr.down()
click.echo("stopped")
except StackError as e:
logger.error(str(e))
click.secho(str(e), fg="red", err=True)
sys.exit(1)
@cli.command("status", help=ch.CMD_DAEMON_STATUS, short_help=ch.CMD_DAEMON_STATUS)
def status_cmd() -> None:
mgr = StackManager()
# Status is what a user runs against an already-up stack, the one case the
# start path's warning never covers (issue #1380).
mgr.warn_if_ports_are_public()
_print_status(mgr)
@cli.command("restart", help=ch.CMD_DAEMON_RESTART, short_help=ch.CMD_DAEMON_RESTART)
def restart_cmd() -> None:
mgr = StackManager()
try:
mgr.restart()
mgr.wait_healthy()
_print_status(mgr)
except StackError as e:
logger.error(str(e))
click.secho(str(e), fg="red", err=True)
sys.exit(1)
@cli.command("logs", help=ch.CMD_DAEMON_LOGS, short_help=ch.CMD_DAEMON_LOGS)
@click.option("--follow", "-f", is_flag=True, help=ch.HELP_DAEMON_LOGS_FOLLOW)
@click.option("--service", "-s", default=None, help=ch.HELP_DAEMON_LOGS_SERVICE)
def logs_cmd(follow: bool, service: str | None) -> None:
mgr = StackManager()
try:
rc = mgr.logs(service=service, follow=follow)
if rc != 0:
sys.exit(rc)
except StackError as e:
logger.error(str(e))
click.secho(str(e), fg="red", err=True)
sys.exit(1)