Mercurial > libervia-backend

comparison libervia/cli/cmd_list.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_list.py@26b7ed2817da
children
comparison
equal deleted inserted replaced
4074:26b7ed2817da 4075:47401850dec6
1 #!/usr/bin/env python3
2
3
4 # Libervia CLI
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20
21 import json
22 import os
23 from libervia.backend.core.i18n import _
24 from libervia.backend.tools.common import data_format
25 from libervia.cli import common
26 from libervia.cli.constants import Const as C
27 from . import base
28
29 __commands__ = ["List"]
30
31 FIELDS_MAP = "mapping"
32
33
34 class Get(base.CommandBase):
35 def __init__(self, host):
36 base.CommandBase.__init__(
37 self,
38 host,
39 "get",
40 use_verbose=True,
41 use_pubsub=True,
42 pubsub_flags={C.MULTI_ITEMS},
43 pubsub_defaults={"service": _("auto"), "node": _("auto")},
44 use_output=C.OUTPUT_LIST_XMLUI,
45 help=_("get lists"),
46 )
47
48 def add_parser_options(self):
49 pass
50
51 async def start(self):
52 await common.fill_well_known_uri(self, os.getcwd(), "tickets", meta_map={})
53 try:
54 lists_data = data_format.deserialise(
55 await self.host.bridge.list_get(
56 self.args.service,
57 self.args.node,
58 self.args.max,
59 self.args.items,
60 "",
61 self.get_pubsub_extra(),
62 self.profile,
63 ),
64 type_check=list,
65 )
66 except Exception as e:
67 self.disp(f"can't get lists: {e}", error=True)
68 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
69 else:
70 await self.output(lists_data[0])
71 self.host.quit(C.EXIT_OK)
72
73
74 class Set(base.CommandBase):
75 def __init__(self, host):
76 base.CommandBase.__init__(
77 self,
78 host,
79 "set",
80 use_pubsub=True,
81 pubsub_defaults={"service": _("auto"), "node": _("auto")},
82 help=_("set a list item"),
83 )
84
85 def add_parser_options(self):
86 self.parser.add_argument(
87 "-f",
88 "--field",
89 action="append",
90 nargs="+",
91 dest="fields",
92 required=True,
93 metavar=("NAME", "VALUES"),
94 help=_("field(s) to set (required)"),
95 )
96 self.parser.add_argument(
97 "-U",
98 "--update",
99 choices=("auto", "true", "false"),
100 default="auto",
101 help=_("update existing item instead of replacing it (DEFAULT: auto)"),
102 )
103 self.parser.add_argument(
104 "item",
105 nargs="?",
106 default="",
107 help=_("id, URL of the item to update, or nothing for new item"),
108 )
109
110 async def start(self):
111 await common.fill_well_known_uri(self, os.getcwd(), "tickets", meta_map={})
112 if self.args.update == "auto":
113 # we update if we have a item id specified
114 update = bool(self.args.item)
115 else:
116 update = C.bool(self.args.update)
117
118 values = {}
119
120 for field_data in self.args.fields:
121 values.setdefault(field_data[0], []).extend(field_data[1:])
122
123 extra = {"update": update}
124
125 try:
126 item_id = await self.host.bridge.list_set(
127 self.args.service,
128 self.args.node,
129 values,
130 "",
131 self.args.item,
132 data_format.serialise(extra),
133 self.profile,
134 )
135 except Exception as e:
136 self.disp(f"can't set list item: {e}", error=True)
137 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
138 else:
139 self.disp(f"item {str(item_id or self.args.item)!r} set successfully")
140 self.host.quit(C.EXIT_OK)
141
142
143 class Delete(base.CommandBase):
144 def __init__(self, host):
145 base.CommandBase.__init__(
146 self,
147 host,
148 "delete",
149 use_pubsub=True,
150 pubsub_defaults={"service": _("auto"), "node": _("auto")},
151 help=_("delete a list item"),
152 )
153
154 def add_parser_options(self):
155 self.parser.add_argument(
156 "-f", "--force", action="store_true", help=_("delete without confirmation")
157 )
158 self.parser.add_argument(
159 "-N", "--notify", action="store_true", help=_("notify deletion")
160 )
161 self.parser.add_argument(
162 "item",
163 help=_("id of the item to delete"),
164 )
165
166 async def start(self):
167 await common.fill_well_known_uri(self, os.getcwd(), "tickets", meta_map={})
168 if not self.args.item:
169 self.parser.error(_("You need to specify a list item to delete"))
170 if not self.args.force:
171 message = _("Are you sure to delete list item {item_id} ?").format(
172 item_id=self.args.item
173 )
174 await self.host.confirm_or_quit(message, _("item deletion cancelled"))
175 try:
176 await self.host.bridge.list_delete_item(
177 self.args.service,
178 self.args.node,
179 self.args.item,
180 self.args.notify,
181 self.profile,
182 )
183 except Exception as e:
184 self.disp(_("can't delete item: {e}").format(e=e), error=True)
185 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
186 else:
187 self.disp(_("item {item} has been deleted").format(item=self.args.item))
188 self.host.quit(C.EXIT_OK)
189
190
191 class Import(base.CommandBase):
192 # TODO: factorize with blog/import
193
194 def __init__(self, host):
195 super().__init__(
196 host,
197 "import",
198 use_progress=True,
199 use_verbose=True,
200 help=_("import tickets from external software/dataset"),
201 )
202
203 def add_parser_options(self):
204 self.parser.add_argument(
205 "importer",
206 nargs="?",
207 help=_("importer name, nothing to display importers list"),
208 )
209 self.parser.add_argument(
210 "-o",
211 "--option",
212 action="append",
213 nargs=2,
214 default=[],
215 metavar=("NAME", "VALUE"),
216 help=_("importer specific options (see importer description)"),
217 )
218 self.parser.add_argument(
219 "-m",
220 "--map",
221 action="append",
222 nargs=2,
223 default=[],
224 metavar=("IMPORTED_FIELD", "DEST_FIELD"),
225 help=_(
226 "specified field in import data will be put in dest field (default: use "
227 "same field name, or ignore if it doesn't exist)"
228 ),
229 )
230 self.parser.add_argument(
231 "-s",
232 "--service",
233 default="",
234 metavar="PUBSUB_SERVICE",
235 help=_("PubSub service where the items must be uploaded (default: server)"),
236 )
237 self.parser.add_argument(
238 "-n",
239 "--node",
240 default="",
241 metavar="PUBSUB_NODE",
242 help=_(
243 "PubSub node where the items must be uploaded (default: tickets' "
244 "defaults)"
245 ),
246 )
247 self.parser.add_argument(
248 "location",
249 nargs="?",
250 help=_(
251 "importer data location (see importer description), nothing to show "
252 "importer description"
253 ),
254 )
255
256 async def on_progress_started(self, metadata):
257 self.disp(_("Tickets upload started"), 2)
258
259 async def on_progress_finished(self, metadata):
260 self.disp(_("Tickets uploaded successfully"), 2)
261
262 async def on_progress_error(self, error_msg):
263 self.disp(
264 _("Error while uploading tickets: {error_msg}").format(error_msg=error_msg),
265 error=True,
266 )
267
268 async def start(self):
269 if self.args.location is None:
270 # no location, the list of importer or description is requested
271 for name in ("option", "service", "node"):
272 if getattr(self.args, name):
273 self.parser.error(
274 _(
275 "{name} argument can't be used without location argument"
276 ).format(name=name)
277 )
278 if self.args.importer is None:
279 self.disp(
280 "\n".join(
281 [
282 f"{name}: {desc}"
283 for name, desc in await self.host.bridge.ticketsImportList()
284 ]
285 )
286 )
287 else:
288 try:
289 short_desc, long_desc = await self.host.bridge.ticketsImportDesc(
290 self.args.importer
291 )
292 except Exception as e:
293 self.disp(f"can't get importer description: {e}", error=True)
294 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
295 else:
296 self.disp(f"{name}: {short_desc}\n\n{long_desc}")
297 self.host.quit()
298 else:
299 # we have a location, an import is requested
300
301 if self.args.progress:
302 # we use a custom progress bar template as we want a counter
303 self.pbar_template = [
304 _("Progress: "),
305 ["Percentage"],
306 " ",
307 ["Bar"],
308 " ",
309 ["Counter"],
310 " ",
311 ["ETA"],
312 ]
313
314 options = {key: value for key, value in self.args.option}
315 fields_map = dict(self.args.map)
316 if fields_map:
317 if FIELDS_MAP in options:
318 self.parser.error(
319 _(
320 "fields_map must be specified either preencoded in --option or "
321 "using --map, but not both at the same time"
322 )
323 )
324 options[FIELDS_MAP] = json.dumps(fields_map)
325
326 try:
327 progress_id = await self.host.bridge.ticketsImport(
328 self.args.importer,
329 self.args.location,
330 options,
331 self.args.service,
332 self.args.node,
333 self.profile,
334 )
335 except Exception as e:
336 self.disp(
337 _("Error while trying to import tickets: {e}").format(e=e),
338 error=True,
339 )
340 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
341 else:
342 await self.set_progress_id(progress_id)
343
344
345 class List(base.CommandBase):
346 subcommands = (Get, Set, Delete, Import)
347
348 def __init__(self, host):
349 super(List, self).__init__(
350 host, "list", use_profile=False, help=_("pubsub lists handling")
351 )