libervia-backend: sat_frontends/jp/common.py comparison

comparison sat_frontends/jp/common.py @ 3028:ab2696e34d29

Python 3 port: /!\ this is a huge commit /!\ starting from this commit, SàT is needs Python 3.6+ /!\ SàT maybe be instable or some feature may not work anymore, this will improve with time This patch port backend, bridge and frontends to Python 3. Roughly this has been done this way: - 2to3 tools has been applied (with python 3.7) - all references to python2 have been replaced with python3 (notably shebangs) - fixed files not handled by 2to3 (notably the shell script) - several manual fixes - fixed issues reported by Python 3 that where not handled in Python 2 - replaced "async" with "async_" when needed (it's a reserved word from Python 3.7) - replaced zope's "implements" with @implementer decorator - temporary hack to handle data pickled in database, as str or bytes may be returned, to be checked later - fixed hash comparison for password - removed some code which is not needed anymore with Python 3 - deactivated some code which needs to be checked (notably certificate validation) - tested with jp, fixed reported issues until some basic commands worked - ported Primitivus (after porting dependencies like urwid satext) - more manual fixes

author	Goffi <goffi@goffi.org>
date	Tue, 13 Aug 2019 19:08:41 +0200
parents	d9491cb81726
children	d314d4181f30

comparison

equal deleted inserted replaced

-:ff5bcb12ae60
+:ab2696e34d29
 from sat.core import exceptions
 from sat.tools.common import regex
 from sat.tools.common.ansi import ANSI as A
 from sat.tools.common import uri as xmpp_uri
 from sat.tools import config
-from ConfigParser import NoSectionError, NoOptionError
+from configparser import NoSectionError, NoOptionError
 from collections import namedtuple
 from functools import partial
 import json
 import os
 import os.path
 def ansi_ljust(s, width):
 """ljust method handling ANSI escape codes"""
 cleaned = regex.ansiRemove(s)
-return s + u" " * (width - len(cleaned))
+return s + " " * (width - len(cleaned))
 def ansi_center(s, width):
 """ljust method handling ANSI escape codes"""
 cleaned = regex.ansiRemove(s)
 diff = width - len(cleaned)
 half = diff / 2
-return half * u" " + s + (half + diff % 2) * u" "
+return half * " " + s + (half + diff % 2) * " "
 def ansi_rjust(s, width):
 """ljust method handling ANSI escape codes"""
 cleaned = regex.ansiRemove(s)
-return u" " * (width - len(cleaned)) + s
+return " " * (width - len(cleaned)) + s
 def getTmpDir(sat_conf, cat_dir, sub_dir=None):
 """Return directory used to store temporary files
 # we split the arguments and add the known fields
 # we split arguments first to avoid escaping issues in file names
 return [a.format(**format_kw) for a in shlex.split(cmd_line)]
 except ValueError as e:
 host.disp(
-u"Couldn't parse editor cmd [{cmd}]: {reason}".format(cmd=cmd_line, reason=e)
+"Couldn't parse editor cmd [{cmd}]: {reason}".format(cmd=cmd_line, reason=e)
 )
 return []
 class BaseEdit(object):
-u"""base class for editing commands
+"""base class for editing commands
 This class allows to edit file for PubSub or something else.
 It works with temporary files in SàT local_dir, in a "cat_dir" subdir
 """
 If there are more file in SECURE_UNLINK_DIR than SECURE_UNLINK_MAX,
 older file are deleted
 @param path(str): file to unlink
 """
 if not os.path.isfile(path):
-raise OSError(u"path must link to a regular file")
+raise OSError("path must link to a regular file")
 if not path.startswith(getTmpDir(self.sat_conf, self.cat_dir_str)):
 self.disp(
-u"File {} is not in SàT temporary hierarchy, we do not remove it".format(
+"File {} is not in SàT temporary hierarchy, we do not remove it".format(
-path.decode("utf-8")
+path
 ),
 2,
 )
 return
 # we have 2 files per draft with use_metadata, so we double max
 os.makedirs(backup_dir)
 filename = os.path.basename(path)
 backup_path = os.path.join(backup_dir, filename)
 # we move file to backup dir
 self.host.disp(
-u"Backuping file {src} to {dst}".format(
+"Backuping file {src} to {dst}".format(
-src=path.decode("utf-8"), dst=backup_path.decode("utf-8")
+src=path, dst=backup_path
 ),
 1,
 )
 os.rename(path, backup_path)
 # and if we exceeded the limit, we remove older file
 backup_files = [os.path.join(backup_dir, f) for f in os.listdir(backup_dir)]
 if len(backup_files) > unlink_max:
 backup_files.sort(key=lambda path: os.stat(path).st_mtime)
 for path in backup_files[: len(backup_files) - unlink_max]:
-self.host.disp(u"Purging backup file {}".format(path.decode("utf-8")), 2)
+self.host.disp("Purging backup file {}".format(path), 2)
 os.unlink(path)
 def runEditor(
 self,
 editor_args_opt,
 editor_exit = subprocess.call([editor] + args)
 # edition will now be checked, and data will be sent if it was a success
 if editor_exit != 0:
 self.disp(
-u"Editor exited with an error code, so temporary file has not be deleted, and item is not published.\nYou can find temporary file at {path}".format(
+"Editor exited with an error code, so temporary file has not be deleted, and item is not published.\nYou can find temporary file at {path}".format(
 path=content_file_path
 ),
 error=True,
 )
 else:
 try:
 with open(content_file_path, "rb") as f:
 content = f.read()
 except (OSError, IOError):
 self.disp(
-u"Can read file at {content_path}, have it been deleted?\nCancelling edition".format(
+"Can read file at {content_path}, have it been deleted?\nCancelling edition".format(
 content_path=content_file_path
 ),
 error=True,
 )
 self.host.quit(C.EXIT_NOT_FOUND)
 try:
 with open(meta_file_path, "rb") as f:
 metadata = json.load(f)
 except (OSError, IOError):
 self.disp(
-u"Can read file at {meta_file_path}, have it been deleted?\nCancelling edition".format(
+"Can read file at {meta_file_path}, have it been deleted?\nCancelling edition".format(
 content_path=content_file_path, meta_path=meta_file_path
 ),
 error=True,
 )
 self.host.quit(C.EXIT_NOT_FOUND)
 except ValueError:
 self.disp(
-u"Can't parse metadata, please check it is correct JSON format. Cancelling edition.\n"
+"Can't parse metadata, please check it is correct JSON format. Cancelling edition.\n"
 + "You can find tmp file at {content_path} and temporary meta file at {meta_path}.".format(
 content_path=content_file_path, meta_path=meta_file_path
 ),
 error=True,
 )
 self.host.quit(C.EXIT_DATA_ERROR)
 if self.use_metadata and not metadata.get("publish", True):
 self.disp(
-u'Publication blocked by "publish" key in metadata, cancelling edition.\n\n'
+'Publication blocked by "publish" key in metadata, cancelling edition.\n\n'
 + "temporary file path:\t{content_path}\nmetadata file path:\t{meta_path}".format(
 content_path=content_file_path, meta_path=meta_file_path
 ),
 error=True,
 )
 self.host.quit()
 if len(content) == 0:
-self.disp(u"Content is empty, cancelling the edition")
+self.disp("Content is empty, cancelling the edition")
 if not content_file_path.startswith(
 getTmpDir(self.sat_conf, self.cat_dir_str)
 ):
 self.disp(
-u"File are not in SàT temporary hierarchy, we do not remove them",
+"File are not in SàT temporary hierarchy, we do not remove them",
 2,
 )
 self.host.quit()
-self.disp(u"Deletion of {}".format(content_file_path.decode("utf-8")), 2)
+self.disp("Deletion of {}".format(content_file_path), 2)
 os.unlink(content_file_path)
 if self.use_metadata:
-self.disp(u"Deletion of {}".format(meta_file_path.decode("utf-8")), 2)
+self.disp("Deletion of {}".format(meta_file_path), 2)
 os.unlink(meta_file_path)
 self.host.quit()
 # time to re-check the hash
 elif tmp_ori_hash == hashlib.sha1(content).digest() and (
 not self.use_metadata or meta_ori == metadata
 ):
-self.disp(u"The content has not been modified, cancelling the edition")
+self.disp("The content has not been modified, cancelling the edition")
 self.host.quit()
 else:
 # we can now send the item
 content = content.decode("utf-8-sig")  # we use utf-8-sig to avoid BOM
 else:
 self.publish(content)
 except Exception as e:
 if self.use_metadata:
 self.disp(
-u"Error while sending your item, the temporary files have been kept at {content_path} and {meta_path}: {reason}".format(
+"Error while sending your item, the temporary files have been kept at {content_path} and {meta_path}: {reason}".format(
 content_path=content_file_path,
 meta_path=meta_file_path,
 reason=e,
 ),
 error=True,
 )
 else:
 self.disp(
-u"Error while sending your item, the temporary file has been kept at {content_path}: {reason}".format(
+"Error while sending your item, the temporary file has been kept at {content_path}: {reason}".format(
 content_path=content_file_path, reason=e
 ),
 error=True,
 )
 self.host.quit(1)
 if not os.path.exists(tmp_dir):
 try:
 os.makedirs(tmp_dir)
 except OSError as e:
 self.disp(
-u"Can't create {path} directory: {reason}".format(
+"Can't create {path} directory: {reason}".format(
 path=tmp_dir, reason=e
 ),
 error=True,
 )
 self.host.quit(1)
 text=True,
 )
 return os.fdopen(fd, "w+b"), path
 except OSError as e:
 self.disp(
-u"Can't create temporary file: {reason}".format(reason=e), error=True
+"Can't create temporary file: {reason}".format(reason=e), error=True
 )
 self.host.quit(1)
 def getCurrentFile(self, profile):
 """Get most recently edited file
 for path in glob.glob(os.path.join(tmp_dir, cat_dir_str + "_*"))
 if not path.endswith(METADATA_SUFF)
 ]
 if not available:
 self.disp(
-u"Could not find any content draft in {path}".format(path=tmp_dir),
+"Could not find any content draft in {path}".format(path=tmp_dir),
 error=True,
 )
 self.host.quit(1)
 return max(available, key=lambda path: os.stat(path).st_mtime)
 """return formatted content, metadata (or not if use_metadata is false), and item id"""
 raise NotImplementedError
 def getTmpSuff(self):
 """return suffix used for content file"""
-return u"xml"
+return "xml"
 def getItemPath(self):
 """retrieve item path (i.e. service and node) from item argument
 This method is obviously only useful for edition of PubSub based features
 last_item = self.args.last_item
 if self.args.current:
 # user wants to continue current draft
 content_file_path = self.getCurrentFile(self.profile)
-self.disp(u"Continuing edition of current draft", 2)
+self.disp("Continuing edition of current draft", 2)
 content_file_obj = open(content_file_path, "r+b")
 # we seek at the end of file in case of an item already exist
 # this will write content of the existing item at the end of the draft.
 # This way no data should be lost.
 content_file_obj.seek(0, os.SEEK_END)
 else:
 # we need a temporary file
 content_file_obj, content_file_path = self.getTmpFile()
 if item or last_item:
-self.disp(u"Editing requested published item", 2)
+self.disp("Editing requested published item", 2)
 try:
 if self.use_metadata:
 content, metadata, item = self.getItemData(service, node, item)
 else:
 content, item = self.getItemData(service, node, item)
 except Exception as e:
 # FIXME: ugly but we have not good may to check errors in bridge
-if u"item-not-found" in unicode(e):
+if "item-not-found" in str(e):
 #  item doesn't exist, we create a new one with requested id
 metadata = None
 if last_item:
-self.disp(_(u"no item found at all, we create a new one"), 2)
+self.disp(_("no item found at all, we create a new one"), 2)
 else:
 self.disp(
 _(
-u'item "{item_id}" not found, we create a new item with this id'
+'item "{item_id}" not found, we create a new item with this id'
 ).format(item_id=item),
 2,
 )
 content_file_obj.seek(0)
 else:
-self.disp(u"Error while retrieving item: {}".format(e))
+self.disp("Error while retrieving item: {}".format(e))
 self.host.quit(C.EXIT_ERROR)
 else:
 # item exists, we write content
 if content_file_obj.tell() != 0:
 # we already have a draft,
 # we copy item content after it and add an indicator
 content_file_obj.write("\n*****\n")
 content_file_obj.write(content.encode("utf-8"))
 content_file_obj.seek(0)
 self.disp(
-_(u'item "{item_id}" found, we edit it').format(item_id=item), 2
+_('item "{item_id}" found, we edit it').format(item_id=item), 2
 )
 else:
-self.disp(u"Editing a new item", 2)
+self.disp("Editing a new item", 2)
 if self.use_metadata:
 metadata = None
 if self.use_metadata:
 return service, node, item, content_file_path, content_file_obj, metadata
 new_row = []
 row_data_list = list(row_data)
 for idx, value in enumerate(row_data_list):
 if filters is not None and filters[idx] is not None:
 filter_ = filters[idx]
-if isinstance(filter_, basestring):
+if isinstance(filter_, str):
 col_value = filter_.format(value)
 else:
 try:
 col_value = filter_(value, row_cls(*row_data_list))
 except TypeError:
 col_value = filter_(value)
 # we count size without ANSI code as they will change length of the string
 # when it's mostly style/color changes.
 col_size = len(regex.ansiRemove(col_value))
 else:
-col_value = unicode(value)
+col_value = str(value)
 col_size = len(col_value)
 new_row.append(col_value)
 if size is None:
 self.sizes.append(col_size)
 else:
 self.sizes[idx] = max(self.sizes[idx], col_size)
 if size is None:
 size = len(new_row)
 if headers is not None and len(headers) != size:
-raise exceptions.DataError(u"headers size is not coherent with rows")
+raise exceptions.DataError("headers size is not coherent with rows")
 else:
 if len(new_row) != size:
-raise exceptions.DataError(u"rows size is not coherent")
+raise exceptions.DataError("rows size is not coherent")
 self.rows.append(new_row)
 if not data and headers is not None:
 #  the table is empty, we print headers at their lenght
 self.sizes = [len(h) for h in headers]
 @property
 def string(self):
 if self._buffer is None:
-raise exceptions.InternalError(u"buffer must be used to get a string")
+raise exceptions.InternalError("buffer must be used to get a string")
-return u"\n".join(self._buffer)
+return "\n".join(self._buffer)
 @staticmethod
 def readDictValues(data, keys, defaults=None):
 if defaults is None:
 defaults = {}
 @param defaults(dict[unicode, unicode]): default value to use
 if None, an exception will be raised if not value is found
 """
 if keys is None and headers is not None:
 # FIXME: keys are not needed with OrderedDict,
-raise exceptions.DataError(u"You must specify keys order to used headers")
+raise exceptions.DataError("You must specify keys order to used headers")
 if keys is None:
-keys = data[0].keys()
+keys = list(data[0].keys())
 if headers is None:
 headers = keys
 filters = [filters.get(k) for k in keys]
 return cls(
 host, (cls.readDictValues(d, keys, defaults) for d in data), headers, filters
 )
-def _headers(self, head_sep, headers, sizes, alignment=u"left", style=None):
+def _headers(self, head_sep, headers, sizes, alignment="left", style=None):
 """Render headers
 @param head_sep(unicode): sequence to use as separator
 @param alignment(unicode): how to align, can be left, center or right
 @param style(unicode, iterable[unicode], None): ANSI escape sequences to apply
 @param headers(list[unicode]): headers to show
 @param sizes(list[int]): sizes of columns
 """
 rendered_headers = []
-if isinstance(style, basestring):
+if isinstance(style, str):
 style = [style]
 for idx, header in enumerate(headers):
 size = sizes[idx]
-if alignment == u"left":
+if alignment == "left":
 rendered = header[:size].ljust(size)
-elif alignment == u"center":
+elif alignment == "center":
 rendered = header[:size].center(size)
-elif alignment == u"right":
+elif alignment == "right":
 rendered = header[:size].rjust(size)
 else:
-raise exceptions.InternalError(u"bad alignment argument")
+raise exceptions.InternalError("bad alignment argument")
 if style:
 args = style + [rendered]
 rendered = A.color(*args)
 rendered_headers.append(rendered)
 return head_sep.join(rendered_headers)
 else:
 self.host.disp(data)
 def display(
 self,
-head_alignment=u"left",
+head_alignment="left",
-columns_alignment=u"left",
+columns_alignment="left",
 head_style=None,
 show_header=True,
 show_borders=True,
 hide_cols=None,
-col_sep=u" │ ",
+col_sep=" │ ",
-top_left=u"┌",
+top_left="┌",
-top=u"─",
+top="─",
-top_sep=u"─┬─",
+top_sep="─┬─",
-top_right=u"┐",
+top_right="┐",
-left=u"│",
+left="│",
 right=None,
 head_sep=None,
-head_line=u"┄",
+head_line="┄",
-head_line_left=u"├",
+head_line_left="├",
-head_line_sep=u"┄┼┄",
+head_line_sep="┄┼┄",
-head_line_right=u"┤",
+head_line_right="┤",
-bottom_left=u"└",
+bottom_left="└",
 bottom=None,
-bottom_sep=u"─┴─",
+bottom_sep="─┴─",
-bottom_right=u"┘",
+bottom_right="┘",
 ):
 """Print the table
 @param show_header(bool): True if header need no be shown
 @param show_borders(bool): True if borders need no be shown
 if bottom is None:
 bottom = top
 if bottom_sep is None:
 bottom_sep = col_sep_size * bottom
 if not show_borders:
-left = right = head_line_left = head_line_right = u""
+left = right = head_line_left = head_line_right = ""
 # top border
 if show_borders:
 self._disp(
 top_left + top_sep.join([top * size for size in sizes]) + top_right
 )
 + head_line_sep.join([head_line * size for size in sizes])
 + head_line_right
 )
 # content
-if columns_alignment == u"left":
+if columns_alignment == "left":
 alignment = lambda idx, s: ansi_ljust(s, sizes[idx])
-elif columns_alignment == u"center":
+elif columns_alignment == "center":
 alignment = lambda idx, s: ansi_center(s, sizes[idx])
-elif columns_alignment == u"right":
+elif columns_alignment == "right":
 alignment = lambda idx, s: ansi_rjust(s, sizes[idx])
 else:
-raise exceptions.InternalError(u"bad columns alignment argument")
+raise exceptions.InternalError("bad columns alignment argument")
 for row in self.rows:
 if hide_cols:
 row = [v for idx, v in enumerate(row) if idx not in ignore_idx]
 self._disp(
 #  we return self so string can be used after display (table.display().string)
 return self
 def display_blank(self, **kwargs):
 """Display table without visible borders"""
-kwargs_ = {"col_sep": u" ", "head_line_sep": u" ", "show_borders": False}
+kwargs_ = {"col_sep": " ", "head_line_sep": " ", "show_borders": False}
 kwargs_.update(kwargs)
 return self.display(**kwargs_)
 class URIFinder(object):
 path,
 [key],
 callback=self.URIFindCb,
 errback=partial(
 command.errback,
-msg=_(u"can't find " + key + u" URI: {}"),
+msg=_("can't find " + key + " URI: {}"),
 exit_code=C.EXIT_BRIDGE_ERRBACK,
 ),
 )
 else:
 callback()
 try:
 values = getattr(self.args, key)
 except AttributeError:
 raise exceptions.InternalError(
-u'there is no "{key}" arguments'.format(key=key)
+'there is no "{key}" arguments'.format(key=key)
 )
 else:
 if values is None:
 values = []
 values.extend(json.loads(new_values_json))
 try:
 uri_data = uris_data[self.key]
 except KeyError:
 self.host.disp(
 _(
-u"No {key} URI specified for this project, please specify service and node"
+"No {key} URI specified for this project, please specify service and node"
 ).format(key=self.key),
 error=True,
 )
 self.host.quit(C.EXIT_NOT_FOUND)
 else:
-uri = uri_data[u"uri"]
+uri = uri_data["uri"]
-self.setMetadataList(uri_data, u"labels")
+self.setMetadataList(uri_data, "labels")
 parsed_uri = xmpp_uri.parseXMPPUri(uri)
 try:
-self.args.service = parsed_uri[u"path"]
+self.args.service = parsed_uri["path"]
-self.args.node = parsed_uri[u"node"]
+self.args.node = parsed_uri["node"]
 except KeyError:
-self.host.disp(_(u"Invalid URI found: {uri}").format(uri=uri), error=True)
+self.host.disp(_("Invalid URI found: {uri}").format(uri=uri), error=True)
 self.host.quit(C.EXIT_DATA_ERROR)
 self.callback()

Mercurial > libervia-backend

comparison sat_frontends/jp/common.py @ 3028:ab2696e34d29