Mercurial > libervia-backend

comparison libervia/cli/cmd_application.py @ 4075:47401850dec6

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: rename `libervia.frontends.jp` to `libervia.cli`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 14:54:26 +0200
parents libervia/frontends/jp/cmd_application.py@26b7ed2817da
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4074:26b7ed2817da 4075:47401850dec6
1 #!/usr/bin/env python3
2
3 # Libervia CLI
4 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from . import base
20 from libervia.backend.core.i18n import _
21 from libervia.backend.tools.common import data_format
22 from libervia.cli.constants import Const as C
23
24 __commands__ = ["Application"]
25
26
27 class List(base.CommandBase):
28 """List available applications"""
29
30 def __init__(self, host):
31 super(List, self).__init__(
32 host, "list", use_profile=False, use_output=C.OUTPUT_LIST,
33 help=_("list available applications")
34 )
35
36 def add_parser_options(self):
37 # FIXME: "extend" would be better here, but it's only available from Python 3.8+
38 # so we use "append" until minimum version of Python is raised.
39 self.parser.add_argument(
40 "-f",
41 "--filter",
42 dest="filters",
43 action="append",
44 choices=["available", "running"],
45 help=_("show applications with this status"),
46 )
47
48 async def start(self):
49
50 # FIXME: this is only needed because we can't use "extend" in
51 # add_parser_options, see note there
52 if self.args.filters:
53 self.args.filters = list(set(self.args.filters))
54 else:
55 self.args.filters = ['available']
56
57 try:
58 found_apps = await self.host.bridge.applications_list(self.args.filters)
59 except Exception as e:
60 self.disp(f"can't get applications list: {e}", error=True)
61 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
62 else:
63 await self.output(found_apps)
64 self.host.quit()
65
66
67 class Start(base.CommandBase):
68 """Start an application"""
69
70 def __init__(self, host):
71 super(Start, self).__init__(
72 host, "start", use_profile=False, help=_("start an application")
73 )
74
75 def add_parser_options(self):
76 self.parser.add_argument(
77 "name",
78 help=_("name of the application to start"),
79 )
80
81 async def start(self):
82 try:
83 await self.host.bridge.application_start(
84 self.args.name,
85 "",
86 )
87 except Exception as e:
88 self.disp(f"can't start {self.args.name}: {e}", error=True)
89 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
90 else:
91 self.host.quit()
92
93
94 class Stop(base.CommandBase):
95
96 def __init__(self, host):
97 super(Stop, self).__init__(
98 host, "stop", use_profile=False, help=_("stop a running application")
99 )
100
101 def add_parser_options(self):
102 id_group = self.parser.add_mutually_exclusive_group(required=True)
103 id_group.add_argument(
104 "name",
105 nargs="?",
106 help=_("name of the application to stop"),
107 )
108 id_group.add_argument(
109 "-i",
110 "--id",
111 help=_("identifier of the instance to stop"),
112 )
113
114 async def start(self):
115 try:
116 if self.args.name is not None:
117 args = [self.args.name, "name"]
118 else:
119 args = [self.args.id, "instance"]
120 await self.host.bridge.application_stop(
121 *args,
122 "",
123 )
124 except Exception as e:
125 if self.args.name is not None:
126 self.disp(
127 f"can't stop application {self.args.name!r}: {e}", error=True)
128 else:
129 self.disp(
130 f"can't stop application instance with id {self.args.id!r}: {e}",
131 error=True)
132 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
133 else:
134 self.host.quit()
135
136
137 class Exposed(base.CommandBase):
138
139 def __init__(self, host):
140 super(Exposed, self).__init__(
141 host, "exposed", use_profile=False, use_output=C.OUTPUT_DICT,
142 help=_("show data exposed by a running application")
143 )
144
145 def add_parser_options(self):
146 id_group = self.parser.add_mutually_exclusive_group(required=True)
147 id_group.add_argument(
148 "name",
149 nargs="?",
150 help=_("name of the application to check"),
151 )
152 id_group.add_argument(
153 "-i",
154 "--id",
155 help=_("identifier of the instance to check"),
156 )
157
158 async def start(self):
159 try:
160 if self.args.name is not None:
161 args = [self.args.name, "name"]
162 else:
163 args = [self.args.id, "instance"]
164 exposed_data_raw = await self.host.bridge.application_exposed_get(
165 *args,
166 "",
167 )
168 except Exception as e:
169 if self.args.name is not None:
170 self.disp(
171 f"can't get values exposed from application {self.args.name!r}: {e}",
172 error=True)
173 else:
174 self.disp(
175 f"can't values exposed from application instance with id {self.args.id!r}: {e}",
176 error=True)
177 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
178 else:
179 exposed_data = data_format.deserialise(exposed_data_raw)
180 await self.output(exposed_data)
181 self.host.quit()
182
183
184 class Application(base.CommandBase):
185 subcommands = (List, Start, Stop, Exposed)
186
187 def __init__(self, host):
188 super(Application, self).__init__(
189 host, "application", use_profile=False, help=_("manage applications"),
190 aliases=['app'],
191 )