libervia-backend: sat_frontends/jp/cmd

comparison sat_frontends/jp/cmd_file.py @ 4037:524856bd7b19

massive refactoring to switch from camelCase to snake_case: historically, Libervia (SàT before) was using camelCase as allowed by PEP8 when using a pre-PEP8 code, to use the same coding style as in Twisted. However, snake_case is more readable and it's better to follow PEP8 best practices, so it has been decided to move on full snake_case. Because Libervia has a huge codebase, this ended with a ugly mix of camelCase and snake_case. To fix that, this patch does a big refactoring by renaming every function and method (including bridge) that are not coming from Twisted or Wokkel, to use fully snake_case. This is a massive change, and may result in some bugs.

author	Goffi <goffi@goffi.org>
date	Sat, 08 Apr 2023 13:54:42 +0200
parents	4c3361e2bf55
children	2594e1951cf7

comparison

equal deleted inserted replaced

-:c4464d7ae97b
+:524856bd7b19
 "--encrypt",
 action="store_true",
 help=_("end-to-end encrypt the file transfer")
 )
-async def onProgressStarted(self, metadata):
+async def on_progress_started(self, metadata):
 self.disp(_("File copy started"), 2)
-async def onProgressFinished(self, metadata):
+async def on_progress_finished(self, metadata):
 self.disp(_("File sent successfully"), 2)
-async def onProgressError(self, error_msg):
+async def on_progress_error(self, error_msg):
 if error_msg == C.PROGRESS_ERROR_DECLINED:
 self.disp(_("The file has been refused by your contact"))
 else:
 self.disp(_("Error while sending file: {}").format(error_msg), error=True)
-async def gotId(self, data, file_):
+async def got_id(self, data, file_):
 """Called when a progress id has been received
 @param pid(unicode): progress id
 @param file_(str): file path
 """
 if self.args.encrypt:
 extra["encrypted"] = True
 if self.args.bz2:
 with tempfile.NamedTemporaryFile("wb", delete=False) as buf:
-self.host.addOnQuitCallback(os.unlink, buf.name)
+self.host.add_on_quit_callback(os.unlink, buf.name)
 self.disp(_("bz2 is an experimental option, use with caution"))
 # FIXME: check free space
 self.disp(_("Starting compression, please wait..."))
 sys.stdout.flush()
 bz2 = tarfile.open(mode="w:bz2", fileobj=buf)
 bz2.add(file_)
 bz2.close()
 self.disp(_("Done !"), 1)
 try:
-send_data = await self.host.bridge.fileSend(
+send_data = await self.host.bridge.file_send(
 self.args.jid,
 buf.name,
 self.args.name or archive_name,
 "",
 data_format.serialise(extra),
 )
 except Exception as e:
 self.disp(f"can't send file: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-await self.gotId(send_data, file_)
+await self.got_id(send_data, file_)
 else:
 for file_ in self.args.files:
 path = os.path.abspath(file_)
 try:
-send_data = await self.host.bridge.fileSend(
+send_data = await self.host.bridge.file_send(
 self.args.jid,
 path,
 self.args.name,
 "",
 data_format.serialise(extra),
 )
 except Exception as e:
 self.disp(f"can't send file {file_!r}: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-await self.gotId(send_data, file_)
+await self.got_id(send_data, file_)
 class Request(base.CommandBase):
 def __init__(self, host):
 super(Request, self).__init__(
 "--force",
 action="store_true",
 help=_("overwrite existing file without confirmation"),
 )
-async def onProgressStarted(self, metadata):
+async def on_progress_started(self, metadata):
 self.disp(_("File copy started"), 2)
-async def onProgressFinished(self, metadata):
+async def on_progress_finished(self, metadata):
 self.disp(_("File received successfully"), 2)
-async def onProgressError(self, error_msg):
+async def on_progress_error(self, error_msg):
 if error_msg == C.PROGRESS_ERROR_DECLINED:
 self.disp(_("The file request has been refused"))
 else:
 self.disp(_("Error while requesting file: {}").format(error_msg), error=True)
 if os.path.exists(path) and not self.args.force:
 message = _("File {path} already exists! Do you want to overwrite?").format(
 path=path
 )
-await self.host.confirmOrQuit(message, _("file request cancelled"))
+await self.host.confirm_or_quit(message, _("file request cancelled"))
 self.full_dest_jid = await self.host.get_full_jid(self.args.jid)
 extra = {}
 if self.args.path:
 extra["path"] = self.args.path
 if self.args.namespace:
 extra["namespace"] = self.args.namespace
 try:
-progress_id = await self.host.bridge.fileJingleRequest(
+progress_id = await self.host.bridge.file_jingle_request(
 self.full_dest_jid,
 path,
 self.args.name,
 self.args.hash,
 self.args.hash_algo if self.args.hash else "",
 use_verbose=True,
 help=_("wait for a file to be sent by a contact"),
 )
 self._overwrite_refused = False  # True when one overwrite as already been refused
 self.action_callbacks = {
-C.META_TYPE_FILE: self.onFileAction,
+C.META_TYPE_FILE: self.on_file_action,
-C.META_TYPE_OVERWRITE: self.onOverwriteAction,
+C.META_TYPE_OVERWRITE: self.on_overwrite_action,
-C.META_TYPE_NOT_IN_ROSTER_LEAK: self.onNotInRosterAction,
+C.META_TYPE_NOT_IN_ROSTER_LEAK: self.on_not_in_roster_action,
 }
 def add_parser_options(self):
 self.parser.add_argument(
 "jids",
 default=".",
 metavar="DIR",
 help=_("destination path (default: working directory)"),
 )
-async def onProgressStarted(self, metadata):
+async def on_progress_started(self, metadata):
 self.disp(_("File copy started"), 2)
-async def onProgressFinished(self, metadata):
+async def on_progress_finished(self, metadata):
 self.disp(_("File received successfully"), 2)
 if metadata.get("hash_verified", False):
 try:
 self.disp(
 _("hash checked: {metadata['hash_algo']}:{metadata['hash']}"), 1
 except KeyError:
 self.disp(_("hash is checked but hash value is missing", 1), error=True)
 else:
 self.disp(_("hash can't be verified"), 1)
-async def onProgressError(self, e):
+async def on_progress_error(self, e):
 self.disp(_("Error while receiving file: {e}").format(e=e), error=True)
-def getXmluiId(self, action_data):
+def get_xmlui_id(self, action_data):
 # FIXME: we temporarily use ElementTree, but a real XMLUI managing module
 #        should be available in the futur
 # TODO: XMLUI module
 try:
 xml_ui = action_data["xmlui"]
 xmlui_id = ui.get("submit")
 if not xmlui_id:
 self.disp(_("Invalid XMLUI received"), error=True)
 return xmlui_id
-async def onFileAction(self, action_data, action_id, security_limit, profile):
+async def on_file_action(self, action_data, action_id, security_limit, profile):
-xmlui_id = self.getXmluiId(action_data)
+xmlui_id = self.get_xmlui_id(action_data)
 if xmlui_id is None:
-return self.host.quitFromSignal(1)
+return self.host.quit_from_signal(1)
 try:
 from_jid = jid.JID(action_data["meta_from_jid"])
 except KeyError:
 self.disp(_("Ignoring action without from_jid data"), 1)
 return
 return
 if not self.bare_jids or from_jid.bare in self.bare_jids:
 if self._overwrite_refused:
 self.disp(_("File refused because overwrite is needed"), error=True)
-await self.host.bridge.launchAction(
+await self.host.bridge.action_launch(
 xmlui_id, {"cancelled": C.BOOL_TRUE}, profile_key=profile
 )
-return self.host.quitFromSignal(2)
+return self.host.quit_from_signal(2)
 await self.set_progress_id(progress_id)
 xmlui_data = {"path": self.path}
-await self.host.bridge.launchAction(xmlui_id, xmlui_data, profile_key=profile)
+await self.host.bridge.action_launch(xmlui_id, xmlui_data, profile_key=profile)
-async def onOverwriteAction(self, action_data, action_id, security_limit, profile):
+async def on_overwrite_action(self, action_data, action_id, security_limit, profile):
-xmlui_id = self.getXmluiId(action_data)
+xmlui_id = self.get_xmlui_id(action_data)
 if xmlui_id is None:
-return self.host.quitFromSignal(1)
+return self.host.quit_from_signal(1)
 try:
 progress_id = action_data["meta_progress_id"]
 except KeyError:
 self.disp(_("ignoring action without progress id"), 1)
 return
 self.disp(_("Overwrite accepted"), 2)
 else:
 self.disp(_("Refused to overwrite"), 2)
 self._overwrite_refused = True
-xmlui_data = {"answer": C.boolConst(self.args.force)}
+xmlui_data = {"answer": C.bool_const(self.args.force)}
-await self.host.bridge.launchAction(xmlui_id, xmlui_data, profile_key=profile)
+await self.host.bridge.action_launch(xmlui_id, xmlui_data, profile_key=profile)
-async def onNotInRosterAction(self, action_data, action_id, security_limit, profile):
+async def on_not_in_roster_action(self, action_data, action_id, security_limit, profile):
-xmlui_id = self.getXmluiId(action_data)
+xmlui_id = self.get_xmlui_id(action_data)
 if xmlui_id is None:
-return self.host.quitFromSignal(1)
+return self.host.quit_from_signal(1)
 try:
 from_jid = jid.JID(action_data["meta_from_jid"])
 except ValueError:
 self.disp(
 _('invalid "from_jid" value received, ignoring: {value}').format(
 self.disp(_("Sender confirmed because she or he is explicitly expected"), 1)
 else:
 xmlui = xmlui_manager.create(self.host, action_data["xmlui"])
 confirmed = await self.host.confirm(xmlui.dlg.message)
-xmlui_data = {"answer": C.boolConst(confirmed)}
+xmlui_data = {"answer": C.bool_const(confirmed)}
-await self.host.bridge.launchAction(xmlui_id, xmlui_data, profile_key=profile)
+await self.host.bridge.action_launch(xmlui_id, xmlui_data, profile_key=profile)
 if not confirmed and not self.args.multiple:
 self.disp(_("Session refused for {from_jid}").format(from_jid=from_jid))
-self.host.quitFromSignal(0)
+self.host.quit_from_signal(0)
 async def start(self):
 self.bare_jids = [jid.JID(jid_).bare for jid_ in self.args.jids]
 self.path = os.path.abspath(self.args.path)
 if not os.path.isdir(self.path):
 self.parser.add_argument(
 "attachment", type=str,
 help=_("URI of the file to retrieve or JSON of the whole attachment")
 )
-async def onProgressStarted(self, metadata):
+async def on_progress_started(self, metadata):
 self.disp(_("File download started"), 2)
-async def onProgressFinished(self, metadata):
+async def on_progress_finished(self, metadata):
 self.disp(_("File downloaded successfully"), 2)
-async def onProgressError(self, error_msg):
+async def on_progress_error(self, error_msg):
 self.disp(_("Error while downloading file: {}").format(error_msg), error=True)
-async def gotId(self, data):
+async def got_id(self, data):
 """Called when a progress id has been received"""
 try:
 await self.set_progress_id(data["progress"])
 except KeyError:
 if "xmlui" in data:
 dest_file = Path(dest_file).expanduser().resolve()
 if dest_file.exists() and not self.args.force:
 message = _("File {path} already exists! Do you want to overwrite?").format(
 path=dest_file
 )
-await self.host.confirmOrQuit(message, _("file download cancelled"))
+await self.host.confirm_or_quit(message, _("file download cancelled"))
 options = {}
 try:
-download_data_s = await self.host.bridge.fileDownload(
+download_data_s = await self.host.bridge.file_download(
 data_format.serialise(attachment),
 str(dest_file),
 data_format.serialise(options),
 self.profile,
 )
 download_data = data_format.deserialise(download_data_s)
 except Exception as e:
 self.disp(f"error while trying to download a file: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-await self.gotId(download_data)
+await self.got_id(download_data)
 class Upload(base.CommandBase):
 def __init__(self, host):
 super(Upload, self).__init__(
 "--ignore-tls-errors",
 action="store_true",
 help=_(r"ignore invalide TLS certificate (/!\ Dangerous /!\)"),
 )
-async def onProgressStarted(self, metadata):
+async def on_progress_started(self, metadata):
 self.disp(_("File upload started"), 2)
-async def onProgressFinished(self, metadata):
+async def on_progress_finished(self, metadata):
 self.disp(_("File uploaded successfully"), 2)
 try:
 url = metadata["url"]
 except KeyError:
 self.disp("download URL not found in metadata")
 else:
 self.disp(_("URL to retrieve the file:"), 1)
 # XXX: url is displayed alone on a line to make parsing easier
 self.disp(url)
-async def onProgressError(self, error_msg):
+async def on_progress_error(self, error_msg):
 self.disp(_("Error while uploading file: {}").format(error_msg), error=True)
-async def gotId(self, data, file_):
+async def got_id(self, data, file_):
 """Called when a progress id has been received
 @param pid(unicode): progress id
 @param file_(str): file path
 """
 if self.args.encrypt:
 options["encryption"] = C.ENC_AES_GCM
 path = os.path.abspath(file_)
 try:
-upload_data = await self.host.bridge.fileUpload(
+upload_data = await self.host.bridge.file_upload(
 path,
 "",
 self.full_dest_jid,
 data_format.serialise(options),
 self.profile,
 )
 except Exception as e:
 self.disp(f"error while trying to upload a file: {e}", error=True)
 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
 else:
-await self.gotId(upload_data, file_)
+await self.got_id(upload_data, file_)
 class ShareAffiliationsSet(base.CommandBase):
 def __init__(self, host):
 super(ShareAffiliationsSet, self).__init__(
 )
 async def start(self):
 affiliations = dict(self.args.affiliations)
 try:
-affiliations = await self.host.bridge.FISAffiliationsSet(
+affiliations = await self.host.bridge.fis_affiliations_set(
 self.args.jid,
 self.args.namespace,
 self.args.path,
 affiliations,
 self.profile,
 help=_("jid of sharing entity"),
 )
 async def start(self):
 try:
-affiliations = await self.host.bridge.FISAffiliationsGet(
+affiliations = await self.host.bridge.fis_affiliations_get(
 self.args.jid,
 self.args.namespace,
 self.args.path,
 self.profile,
 )
 )
 async def start(self):
 configuration = dict(self.args.fields)
 try:
-configuration = await self.host.bridge.FISConfigurationSet(
+configuration = await self.host.bridge.fis_configuration_set(
 self.args.jid,
 self.args.namespace,
 self.args.path,
 configuration,
 self.profile,
 help=_("jid of sharing entity"),
 )
 async def start(self):
 try:
-configuration = await self.host.bridge.FISConfigurationGet(
+configuration = await self.host.bridge.fis_configuration_get(
 self.args.jid,
 self.args.namespace,
 self.args.path,
 self.profile,
 )
 return name
 def _size_filter(self, size, row):
 if not size:
 return ""
-return A.color(A.BOLD, utils.getHumanSize(size))
+return A.color(A.BOLD, utils.get_human_size(size))
 def default_output(self, files_data):
 """display files a way similar to ls"""
 files_data.sort(key=lambda d: d["name"].lower())
 show_header = False
 keys = headers = ("name", "type", "size")
 elif self.verbosity > 1:
 show_header = True
 keys = ("name", "type", "size", "file_hash")
 headers = ("name", "type", "size", "hash")
-table = common.Table.fromListDict(
+table = common.Table.from_list_dict(
 self.host,
 files_data,
 keys=keys,
 headers=headers,
 filters={"name": self._name_filter, "size": self._size_filter},
 )
 table.display_blank(show_header=show_header, hide_cols=["type"])
 async def start(self):
 try:
-files_data = await self.host.bridge.FISList(
+files_data = await self.host.bridge.fis_list(
 self.args.jid,
 self.args.path,
 {},
 self.profile,
 )
 if jids:
 access = {"read": {"type": "whitelist", "jids": jids}}
 else:
 access = {}
 try:
-name = await self.host.bridge.FISSharePath(
+name = await self.host.bridge.fis_share_path(
 self.args.name,
 self.path,
 json.dumps(access, ensure_ascii=False),
 self.profile,
 )
 if not self.args.thumbnail.startswith("http"):
 self.parser.error(_("only http(s) links are allowed with --thumbnail"))
 else:
 extra["thumb_url"] = self.args.thumbnail
 try:
-await self.host.bridge.FISInvite(
+await self.host.bridge.fis_invite(
 self.args.jid,
 self.args.service,
 self.args.type,
 self.args.namespace,
 self.path,

Mercurial > libervia-backend

comparison sat_frontends/jp/cmd_file.py @ 4037:524856bd7b19