libervia-backend: sat_frontends/jp/cmd

comparison sat_frontends/jp/cmd_pubsub.py @ 3568:04283582966f

core, frontends: fix invalid translatable strings. Some f-strings where used in translatable text, this has been fixed by using explicit `format()` call (using a script based on `tokenize`). As tokenize messes with spaces, a reformating tool (`black`) has been applied to some files afterwards.

author	Goffi <goffi@goffi.org>
date	Mon, 14 Jun 2021 18:35:12 +0200
parents	62f490eff51c
children	5f65f4e9f8cb

comparison

equal deleted inserted replaced

-:a240748ed686
+:04283582966f
 self.args.service,
 self.args.node,
 self.profile,
 )
 except BridgeException as e:
-if e.condition == 'item-not-found':
+if e.condition == "item-not-found":
 self.disp(
 f"The node {self.args.node} doesn't exist on {self.args.service}",
-error=True
+error=True,
 )
 self.host.quit(C.EXIT_NOT_FOUND)
 else:
 self.disp(f"can't get node configuration: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 self.args.node,
 options,
 self.profile,
 )
 except Exception as e:
-self.disp(msg=_(f"can't create node: {e}"), error=True)
+self.disp(msg=_("can't create node: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 if self.host.verbosity:
 announce = _("node created successfully: ")
 else:
 self.disp(announce + node_id)
 self.host.quit()
 class NodePurge(base.CommandBase):
 def __init__(self, host):
 super(NodePurge, self).__init__(
 host,
 "purge",
 use_pubsub=True,
 async def start(self):
 if not self.args.force:
 if not self.args.service:
 message = _(
-f"Are you sure to purge PEP node [{self.args.node}]? This will "
+"Are you sure to purge PEP node [{node}]? This will "
-f"delete ALL items from it!")
+"delete ALL items from it!"
+).format(node=self.args.node)
 else:
 message = _(
-f"Are you sure to delete node [{self.args.node}] on service "
+"Are you sure to delete node [{node}] on service "
-f"[{self.args.service}]? This will delete ALL items from it!")
+"[{service}]? This will delete ALL items from it!"
+).format(node=self.args.node, service=self.args.service)
 await self.host.confirmOrQuit(message, _("node purge cancelled"))
 try:
 await self.host.bridge.psNodePurge(
 self.args.service,
 self.args.node,
 self.profile,
 )
 except Exception as e:
-self.disp(msg=_(f"can't purge node: {e}"), error=True)
+self.disp(msg=_("can't purge node: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-self.disp(_(f"node [{self.args.node}] purged successfully"))
+self.disp(_("node [{node}] purged successfully").format(node=self.args.node))
 self.host.quit()
 class NodeDelete(base.CommandBase):
 def __init__(self, host):
 )
 async def start(self):
 if not self.args.force:
 if not self.args.service:
-message = _(f"Are you sure to delete PEP node [{self.args.node}] ?")
+message = _("Are you sure to delete PEP node [{node}] ?").format(
+node=self.args.node
+)
 else:
-message = _(f"Are you sure to delete node [{self.args.node}] on "
+message = _(
-f"service [{self.args.service}]?")
+"Are you sure to delete node [{node}] on " "service [{service}]?"
+).format(node=self.args.node, service=self.args.service)
 await self.host.confirmOrQuit(message, _("node deletion cancelled"))
 try:
 await self.host.bridge.psNodeDelete(
 self.args.service,
 )
 except Exception as e:
 self.disp(f"can't delete node: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-self.disp(_(f"node [{self.args.node}] deleted successfully"))
+self.disp(_("node [{node}] deleted successfully").format(node=self.args.node))
 self.host.quit()
 class NodeSet(base.CommandBase):
 def __init__(self, host):
 self.disp(_("node configuration successful"), 1)
 self.host.quit()
 class NodeImport(base.CommandBase):
 def __init__(self, host):
 super(NodeImport, self).__init__(
 host,
 "import",
 use_pubsub=True,
 help=_("do a pubsub admin request, needed to change publisher"),
 )
 self.parser.add_argument(
 "import_file",
 type=argparse.FileType(),
-help=_("path to the XML file with data to import. The file must contain "
+help=_(
-"whole XML of each item to import."),
+"path to the XML file with data to import. The file must contain "
-)
+"whole XML of each item to import."
+),
-async def start(self):
+)
-try:
-element, etree = xml_tools.etreeParse(self, self.args.import_file,
+async def start(self):
-reraise=True)
+try:
+element, etree = xml_tools.etreeParse(
+self, self.args.import_file, reraise=True
+)
 except Exception as e:
 from lxml.etree import XMLSyntaxError
 if isinstance(e, XMLSyntaxError) and e.code == 5:
 # we have extra content, this probaby means that item are not wrapped
 # so we wrap them here and try again
 self.args.import_file.seek(0)
 xml_buf = "<import>" + self.args.import_file.read() + "</import>"
 element, etree = xml_tools.etreeParse(self, xml_buf)
 # we reverse element as we expect to have most recently published element first
 # TODO: make this more explicit and add an option
 element[:] = reversed(element)
-if not all([i.tag == '{http://jabber.org/protocol/pubsub}item' for i in element]):
+if not all([i.tag == "{http://jabber.org/protocol/pubsub}item" for i in element]):
 self.disp(
 _("You are not using list of pubsub items, we can't import this file"),
-error=True)
+error=True,
+)
 self.host.quit(C.EXIT_DATA_ERROR)
 return
 items = [etree.tostring(i, encoding="unicode") for i in element]
 if self.args.admin:
 method = self.host.bridge.psAdminItemsSend
 else:
-self.disp(_("Items are imported without using admin mode, publisher can't "
+self.disp(
-"be changed"))
+_(
+"Items are imported without using admin mode, publisher can't "
+"be changed"
+)
+)
 method = self.host.bridge.psItemsSend
 try:
 items_ids = await method(
 self.args.service,
 except Exception as e:
 self.disp(f"can't send items: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 if items_ids:
-self.disp(_('items published with id(s) {items_ids}').format(
+self.disp(
-items_ids=', '.join(items_ids)))
+_("items published with id(s) {items_ids}").format(
+items_ids=", ".join(items_ids)
+)
+)
 else:
-self.disp(_('items published'))
+self.disp(_("items published"))
 self.host.quit()
 class NodeAffiliationsGet(base.CommandBase):
 def __init__(self, host):
 async def psSchemaGetCb(self, schema):
 try:
 from lxml import etree
 except ImportError:
-self.disp('lxml module must be installed to use edit, please install it '
+self.disp(
-'with "pip install lxml"',
+"lxml module must be installed to use edit, please install it "
+'with "pip install lxml"',
 error=True,
 )
 self.host.quit(1)
 content_file_obj, content_file_path = self.getTmpFile()
 schema = schema.strip()
 schema_elt = etree.fromstring(schema, parser)
 content_file_obj.write(
 etree.tostring(schema_elt, encoding="utf-8", pretty_print=True)
 )
 content_file_obj.seek(0)
-await self.runEditor("pubsub_schema_editor_args", content_file_path, content_file_obj)
+await self.runEditor(
+"pubsub_schema_editor_args", content_file_path, content_file_obj
+)
 async def start(self):
 try:
 schema = await self.host.bridge.psSchemaGet(
 self.args.service,
 self.args.node,
 self.profile,
 )
 except BridgeException as e:
-if e.condition == 'item-not-found' or e.classname=="NotFound":
+if e.condition == "item-not-found" or e.classname == "NotFound":
 schema = ""
 else:
 self.disp(f"can't edit schema: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 self.args.service,
 self.args.node,
 self.profile,
 )
 except BridgeException as e:
-if e.condition == 'item-not-found' or e.classname=="NotFound":
+if e.condition == "item-not-found" or e.classname == "NotFound":
 schema = None
 else:
 self.disp(f"can't get schema: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 element = xml_tools.getPayload(self, element)
 payload = etree.tostring(element, encoding="unicode")
 extra = {}
 publish_options = NodeCreate.get_config_options(self.args)
 if publish_options:
-extra['publish_options'] = publish_options
+extra["publish_options"] = publish_options
 try:
 published_id = await self.host.bridge.psItemSend(
 self.args.service,
 self.args.node,
 self.args.item,
 data_format.serialise(extra),
 self.profile,
 )
 except Exception as e:
-self.disp(_(f"can't send item: {e}"), error=True)
+self.disp(_("can't send item: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 if published_id:
 if self.args.quiet:
-self.disp(published_id, end='')
+self.disp(published_id, end="")
 else:
 self.disp(f"Item published at {published_id}")
 else:
 self.disp("Item published")
 self.host.quit(C.EXIT_OK)
 self.getPubsubExtra(),
 self.profile,
 )
 )
 except BridgeException as e:
-if e.condition == 'item-not-found' or e.classname=="NotFound":
+if e.condition == "item-not-found" or e.classname == "NotFound":
 self.disp(
 f"The node {self.args.node} doesn't exist on {self.args.service}",
-error=True
+error=True,
 )
 self.host.quit(C.EXIT_NOT_FOUND)
 else:
 self.disp(f"can't get pubsub items: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 except Exception as e:
 self.disp(f"Internal error: {e}", error=True)
 self.host.quit(C.EXIT_INTERNAL_ERROR)
 else:
-await self.output(ps_result['items'])
+await self.output(ps_result["items"])
 self.host.quit(C.EXIT_OK)
 class Delete(base.CommandBase):
 def __init__(self, host):
 self.args.item,
 self.args.notify,
 self.profile,
 )
 except Exception as e:
-self.disp(_(f"can't delete item: {e}"), error=True)
+self.disp(_("can't delete item: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-self.disp(_(f"item {self.args.item} has been deleted"))
+self.disp(_("item {item} has been deleted").format(item=self.args.item))
 self.host.quit(C.EXIT_OK)
 class Edit(base.CommandBase, common.BaseEdit):
 def __init__(self, host):
 async def getItemData(self, service, node, item):
 try:
 from lxml import etree
 except ImportError:
-self.disp('lxml module must be installed to use edit, please install it '
+self.disp(
-'with "pip install lxml"',
+"lxml module must be installed to use edit, please install it "
+'with "pip install lxml"',
 error=True,
 )
 self.host.quit(1)
 items = [item] if item else []
 ps_result = data_format.deserialise(
 await self.host.bridge.psItemsGet(
 service, node, 1, items, "", {}, self.profile
 )
 )
-item_raw = ps_result['items'][0]
+item_raw = ps_result["items"][0]
 parser = etree.XMLParser(remove_blank_text=True, recover=True)
 item_elt = etree.fromstring(item_raw, parser)
 item_id = item_elt.get("id")
 try:
 payload = item_elt[0]
 self.disp(_("Item has not payload"), 1)
 return "", item_id
 return etree.tostring(payload, encoding="unicode", pretty_print=True), item_id
 async def start(self):
-(self.pubsub_service,
+(
-self.pubsub_node,
+self.pubsub_service,
-self.pubsub_item,
+self.pubsub_node,
-content_file_path,
+self.pubsub_item,
-content_file_obj) = await self.getItemPath()
+content_file_path,
+content_file_obj,
+) = await self.getItemPath()
 await self.runEditor("pubsub_editor_args", content_file_path, content_file_obj)
 self.host.quit()
 class Rename(base.CommandBase):
 def __init__(self, host):
 base.CommandBase.__init__(
 self,
 host,
 "rename",
 pubsub_flags={C.NODE, C.SINGLE_ITEM},
 help=_("rename a pubsub item"),
 )
 def add_parser_options(self):
-self.parser.add_argument(
+self.parser.add_argument("new_id", help=_("new item id to use"))
-"new_id",
-help=_("new item id to use")
-)
 async def start(self):
 try:
 await self.host.bridge.psItemRename(
 self.args.service,
 self.args.item,
 self.args.new_id,
 self.profile,
 )
 except Exception as e:
-self.disp(
+self.disp(f"can't rename item: {e}", error=True)
-f"can't rename item: {e}", error=True
-)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 self.disp("Item renamed")
 self.host.quit(C.EXIT_OK)
 self.args.node,
 {},
 self.profile,
 )
 except Exception as e:
-self.disp(_(f"can't subscribe to node: {e}"), error=True)
+self.disp(_("can't subscribe to node: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 self.disp(_("subscription done"), 1)
 if sub_id:
 self.disp(_("subscription id: {sub_id}").format(sub_id=sub_id))
 self.args.service,
 self.args.node,
 self.profile,
 )
 except Exception as e:
-self.disp(_(f"can't unsubscribe from node: {e}"), error=True)
+self.disp(_("can't unsubscribe from node: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 self.disp(_("subscription removed"), 1)
 self.host.quit()
 self.args.service,
 self.args.node,
 self.profile,
 )
 except Exception as e:
-self.disp(_(f"can't retrieve subscriptions: {e}"), error=True)
+self.disp(_("can't retrieve subscriptions: {e}").format(e=e), error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 await self.output(subscriptions)
 self.host.quit()
 self.args.service,
 self.args.node,
 self.profile,
 )
 except Exception as e:
-self.disp(
+self.disp(f"can't get node affiliations: {e}", error=True)
-f"can't get node affiliations: {e}", error=True
-)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 await self.output(affiliations)
 self.host.quit()
 self.parser.add_argument(
 "-D",
 "--max-depth",
 type=int,
 default=0,
-help=_("maximum depth of recursion (will search linked nodes if > 0, "
+help=_(
-"DEFAULT: 0)"),
+"maximum depth of recursion (will search linked nodes if > 0, "
+"DEFAULT: 0)"
+),
 )
 self.parser.add_argument(
 "-M",
 "--node-max",
 type=int,
 default=30,
-help=_("maximum number of items to get per node ({} to get all items, "
+help=_(
-"DEFAULT: 30)".format( C.NO_LIMIT)),
+"maximum number of items to get per node ({} to get all items, "
+"DEFAULT: 30)".format(C.NO_LIMIT)
+),
 )
 self.parser.add_argument(
 "-N",
 "--namespace",
 action="append",
 "--python",
 action="append",
 dest="filters",
 type=filter_python,
 metavar="PYTHON_CODE",
-help=_('Python expression which much return a bool (True to keep item, '
+help=_(
-'False to reject it). "item" is raw text item, "item_xml" is '
+"Python expression which much return a bool (True to keep item, "
-'lxml\'s etree.Element'
+'False to reject it). "item" is raw text item, "item_xml" is '
+"lxml's etree.Element"
 ),
 )
 # filters flags
 flag_case = partial(self.filter_flag, type_="ignore-case")
 def parseXml(self, item):
 try:
 return self.etree.fromstring(item)
 except self.etree.XMLSyntaxError:
 self.disp(
-_("item doesn't looks like XML, you have probably used --only-matching "
+_(
-"somewhere before and we have no more XML"),
+"item doesn't looks like XML, you have probably used --only-matching "
+"somewhere before and we have no more XML"
+),
 error=True,
 )
 self.host.quit(C.EXIT_BAD_ARG)
 def filter(self, item):
 keep = False
 if keep and only_matching:
 # doesn't really make sens to keep a fixed string
 # so we raise an error
 self.host.disp(
-_(
+_("--only-matching used with fixed --text string, are you sure?"),
-"--only-matching used with fixed --text string, are you sure?"
-),
 error=True,
 )
 self.host.quit(C.EXIT_BAD_ARG)
 elif type_ == "regex":
 flags = self.RE_FLAGS
 if item_xml is None:
 item_xml = self.parseXml(item)
 try:
 elts = item_xml.xpath(value, namespaces=self.args.namespace)
 except self.etree.XPathEvalError as e:
-self.disp(
+self.disp(_("can't use xpath: {reason}").format(reason=e), error=True)
-_("can't use xpath: {reason}").format(reason=e), error=True
-)
 self.host.quit(C.EXIT_BAD_ARG)
 keep = bool(elts)
 if keep and only_matching:
 item_xml = elts[0]
 try:
 item = str(item_xml)
 item_xml = None
 elif type_ == "python":
 if item_xml is None:
 item_xml = self.parseXml(item)
-cmd_ns = {
+cmd_ns = {"etree": self.etree, "item": item, "item_xml": item_xml}
-"etree": self.etree,
-"item": item,
-"item_xml": item_xml
-}
 try:
 keep = eval(value, cmd_ns)
 except SyntaxError as e:
 self.disp(str(e), error=True)
 self.host.quit(C.EXIT_BAD_ARG)
 ## flags
 elif type_ == "ignore-case":
 ignore_case = value
 elif type_ == "invert":
 invert = value
 )
 if action == "exec":
 p = await asyncio.create_subprocess_exec(*cmd_args)
 ret = await p.wait()
 else:
-p = await asyncio.create_subprocess_exec(*cmd_args,
+p = await asyncio.create_subprocess_exec(*cmd_args, stdin=subprocess.PIPE)
-stdin=subprocess.PIPE)
 await p.communicate(item.encode(sys.getfilesystemencoding()))
 ret = p.returncode
 if ret != 0:
 self.disp(
 A.color(
 C.A_FAILURE,
-_(f"executed command failed with exit code {ret}"),
+_("executed command failed with exit code {ret}").format(ret=ret),
 )
 )
 async def search(self, ps_result, depth):
 """callback of getItems
 do the requested action, and exit the command when everything is done
 @param items_data(tuple): result of getItems
 @param depth(int): current depth level
 0 for first node, 1 for first children, and so on
 """
-for item in ps_result['items']:
+for item in ps_result["items"]:
 if depth < self.args.max_depth:
 await self.getSubNodes(item, depth)
 keep, item = self.filter(item)
 if not keep:
 continue
 await self.doItemAction(item, ps_result)
 #  we check if we got all getItems results
 self.to_get -= 1
 if self.to_get == 0:
 # yes, we can quit
 self.host.quit()
 assert self.to_get > 0
 if self.args.action in self.EXEC_ACTIONS:
 self.parser.error(_("you need to specify a command to execute"))
 if not self.args.node:
 # TODO: handle get service affiliations when node is not set
 self.parser.error(_("empty node is not handled yet"))
 # to_get is increased on each get and decreased on each answer
 # when it reach 0 again, the command is finished
 self.to_get = 0
 self._etree = None
 if self.args.filters is None:
 self.args.filters = []
 self.args.namespace = dict(
 self.parser.add_argument(
 "-I",
 "--ignore-errors",
 action="store_true",
 help=_(
-"if command return a non zero exit code, ignore the item and continue"),
+"if command return a non zero exit code, ignore the item and continue"
+),
 )
 self.parser.add_argument(
 "-A",
 "--all",
 action="store_true",
-help=_("get all items by looping over all pages using RSM")
+help=_("get all items by looping over all pages using RSM"),
 )
 self.parser.add_argument(
 "command_path",
-help=_("path to the command to use. Will be called repetitivly with an "
+help=_(
-"item as input. Output (full item XML) will be used as new one. "
+"path to the command to use. Will be called repetitivly with an "
-'Return "DELETE" string to delete the item, and "SKIP" to ignore it'),
+"item as input. Output (full item XML) will be used as new one. "
+'Return "DELETE" string to delete the item, and "SKIP" to ignore it'
+),
 )
 async def psItemsSendCb(self, item_ids, metadata):
 if item_ids:
-self.disp(_('items published with ids {item_ids}').format(
+self.disp(
-item_ids=', '.join(item_ids)))
+_("items published with ids {item_ids}").format(
-else:
+item_ids=", ".join(item_ids)
-self.disp(_('items published'))
+)
+)
+else:
+self.disp(_("items published"))
 if self.args.all:
 return await self.handleNextPage(metadata)
 else:
 self.host.quit()
 use to handle --all option
 @param metadata(dict): metadata as returned by psItemsGet
 """
 try:
-last = metadata['rsm']['last']
+last = metadata["rsm"]["last"]
-index = int(metadata['rsm']['index'])
+index = int(metadata["rsm"]["index"])
-count = int(metadata['rsm']['count'])
+count = int(metadata["rsm"]["count"])
 except KeyError:
-self.disp(_("Can't retrieve all items, RSM metadata not available"),
+self.disp(
-error=True)
+_("Can't retrieve all items, RSM metadata not available"), error=True
+)
 self.host.quit(C.EXIT_MISSING_FEATURE)
 except ValueError as e:
-self.disp(_("Can't retrieve all items, bad RSM metadata: {msg}")
+self.disp(
-.format(msg=e), error=True)
+_("Can't retrieve all items, bad RSM metadata: {msg}").format(msg=e),
+error=True,
+)
 self.host.quit(C.EXIT_ERROR)
 if index + self.args.rsm_max >= count:
-self.disp(_('All items transformed'))
+self.disp(_("All items transformed"))
 self.host.quit(0)
-self.disp(_('Retrieving next page ({page_idx}/{page_total})').format(
+self.disp(
-page_idx = int(index/self.args.rsm_max) + 1,
+_("Retrieving next page ({page_idx}/{page_total})").format(
-page_total = int(count/self.args.rsm_max),
+page_idx=int(index / self.args.rsm_max) + 1,
+page_total=int(count / self.args.rsm_max),
 )
 )
 extra = self.getPubsubExtra()
-extra['rsm_after'] = last
+extra["rsm_after"] = last
 try:
 ps_result = await data_format.deserialise(
 self.host.bridge.psItemsGet(
 self.args.service,
 self.args.node,
 extra,
 self.profile,
 )
 )
 except Exception as e:
-self.disp(
+self.disp(f"can't retrieve items: {e}", error=True)
-f"can't retrieve items: {e}", error=True
-)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 await self.psItemsGetCb(ps_result)
 async def psItemsGetCb(self, ps_result):
-encoding = 'utf-8'
+encoding = "utf-8"
 new_items = []
-for item in ps_result['items']:
+for item in ps_result["items"]:
 if self.check_duplicates:
 # this is used when we are not ordering by creation
 # to avoid infinite loop
 item_elt, __ = xml_tools.etreeParse(self, item)
-item_id = item_elt.get('id')
+item_id = item_elt.get("id")
 if item_id in self.items_ids:
-self.disp(_(
+self.disp(
-"Duplicate found on item {item_id}, we have probably handled "
+_(
-"all items.").format(item_id=item_id))
+"Duplicate found on item {item_id}, we have probably handled "
+"all items."
+).format(item_id=item_id)
+)
 self.host.quit()
 self.items_ids.append(item_id)
 # we launch the command to filter the item
 try:
 p = await asyncio.create_subprocess_exec(
-self.args.command_path,
+self.args.command_path, stdin=subprocess.PIPE, stdout=subprocess.PIPE
-stdin=subprocess.PIPE,
+)
-stdout=subprocess.PIPE)
 except OSError as e:
 exit_code = C.EXIT_CMD_NOT_FOUND if e.errno == 2 else C.EXIT_ERROR
 self.disp(f"Can't execute the command: {e}", error=True)
 self.host.quit(exit_code)
 encoding = "utf-8"
 cmd_std_out, cmd_std_err = await p.communicate(item.encode(encoding))
 ret = p.returncode
 if ret != 0:
-self.disp(f"The command returned a non zero status while parsing the "
+self.disp(
-f"following item:\n\n{item}", error=True)
+f"The command returned a non zero status while parsing the "
+f"following item:\n\n{item}",
+error=True,
+)
 if self.args.ignore_errors:
 continue
 else:
 self.host.quit(C.EXIT_CMD_ERROR)
 if cmd_std_err is not None:
-cmd_std_err = cmd_std_err.decode(encoding, errors='ignore')
+cmd_std_err = cmd_std_err.decode(encoding, errors="ignore")
 self.disp(cmd_std_err, error=True)
 cmd_std_out = cmd_std_out.decode(encoding).strip()
 if cmd_std_out == "DELETE":
 item_elt, __ = xml_tools.etreeParse(self, item)
-item_id = item_elt.get('id')
+item_id = item_elt.get("id")
-self.disp(_(f"Deleting item {item_id}"))
+self.disp(_("Deleting item {item_id}").format(item_id=item_id))
 if self.args.apply:
 try:
 await self.host.bridge.psItemRetract(
 self.args.service,
 self.args.node,
 item_id,
 False,
 self.profile,
 )
 except Exception as e:
-self.disp(
+self.disp(f"can't delete item {item_id}: {e}", error=True)
-f"can't delete item {item_id}: {e}", error=True
-)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 continue
 elif cmd_std_out == "SKIP":
 item_elt, __ = xml_tools.etreeParse(self, item)
-item_id = item_elt.get('id')
+item_id = item_elt.get("id")
 self.disp(_("Skipping item {item_id}").format(item_id=item_id))
 continue
 element, etree = xml_tools.etreeParse(self, cmd_std_out)
 # at this point command has been run and we have a etree.Element object
 if element.tag not in ("item", "{http://jabber.org/protocol/pubsub}item"):
-self.disp("your script must return a whole item, this is not:\n{xml}"
+self.disp(
-.format(xml=etree.tostring(element, encoding="unicode")), error=True)
+"your script must return a whole item, this is not:\n{xml}".format(
+xml=etree.tostring(element, encoding="unicode")
+),
+error=True,
+)
 self.host.quit(C.EXIT_DATA_ERROR)
 if not self.args.apply:
 # we have a dry run, we just display filtered items
-serialised = etree.tostring(element, encoding='unicode',
+serialised = etree.tostring(
-pretty_print=True)
+element, encoding="unicode", pretty_print=True
+)
 self.disp(serialised)
 else:
 new_items.append(etree.tostring(element, encoding="unicode"))
 if not self.args.apply:
 async def start(self):
 if self.args.all and self.args.order_by != C.ORDER_BY_CREATION:
 self.check_duplicates = True
 self.items_ids = []
-self.disp(A.color(
+self.disp(
-A.FG_RED, A.BOLD,
+A.color(
-'/!\\ "--all" should be used with "--order-by creation" /!\\\n',
+A.FG_RED,
-A.RESET,
+A.BOLD,
-"We'll update items, so order may change during transformation,\n"
+'/!\\ "--all" should be used with "--order-by creation" /!\\\n',
-"we'll try to mitigate that by stopping on first duplicate,\n"
+A.RESET,
-"but this method is not safe, and some items may be missed.\n---\n"))
+"We'll update items, so order may change during transformation,\n"
+"we'll try to mitigate that by stopping on first duplicate,\n"
+"but this method is not safe, and some items may be missed.\n---\n",
+)
+)
 else:
 self.check_duplicates = False
 try:
 ps_result = data_format.deserialise(
 async def start(self):
 if not self.args.service:
 try:
 jid_ = await self.host.bridge.asyncGetParamA(
-"JabberID",
+"JabberID", "Connection", profile_key=self.args.profile
-"Connection",
-profile_key=self.args.profile
 )
 except Exception as e:
 self.disp(f"can't retrieve jid: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
 )
 except Exception as e:
 self.disp(f"can't delete hook: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-self.disp(_(f"{nb_deleted} hook(s) have been deleted"))
+self.disp(
+_("{nb_deleted} hook(s) have been deleted").format(nb_deleted=nb_deleted)
+)
 self.host.quit()
 class HookList(base.CommandBase):
 def __init__(self, host):

Mercurial > libervia-backend

comparison sat_frontends/jp/cmd_pubsub.py @ 3568:04283582966f