eric: comparison src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py

-:552a790fd9bc
+:00f5aae565a3
 import contextlib
 import os
 import re
 import stat
-from PyQt6.QtCore import QEventLoop, QObject, pyqtSlot
+from PyQt6.QtCore import QByteArray, QEventLoop, QObject, pyqtSlot
 from eric7 import Utilities
 from eric7.RemoteServer.EricRequestCategory import EricRequestCategory
 from eric7.SystemUtilities import FileSystemUtilities
+_RemoteFsCache = {}
+# dictionary containing cached remote file system data keyed by remote path
 class EricServerNotConnectedError(OSError):
 """
 Class defining a special OSError indicating a missing server connection.
 def __init__(self):
 """
 Constructor
 """
-super().__init("Not connected to an 'eric-ide' server.")
+super().__init__("Not connected to an 'eric-ide' server.")
 class EricServerFileSystemInterface(QObject):
 """
 Class implementing the file system interface to the eric-ide server.
 @type bool
 """
 if connected:
 if not bool(self.__serverPathSep):
 self.__serverPathSep = self.__getPathSep()
-else:
-self.__serverPathSep = ""
 def __getPathSep(self):
 """
 Private method to get the path separator of the connected server.
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
-def listdir(self, directory=""):
+def listdir(self, directory="", recursive=False):
 """
 Public method to get a directory listing.
 @param directory directory to be listed. An empty directory means to list
 the eric-ide server current directory. (defaults to "")
 @type str (optional)
+@param recursive flag indicating a recursive listing (defaults to False)
+@type bool (optional)
 @return tuple containing the listed directory, the path separator and the
 directory listing. Each directory listing entry contains a dictionary
 with the relevant data.
 @rtype tuple of (str, str, dict)
 @exception OSError raised in case the server reported an issue
 if self.__serverInterface.isServerConnected():
 self.__serverInterface.sendJson(
 category=EricRequestCategory.FileSystem,
 request="Listdir",
-params={"directory": FileSystemUtilities.plainFileName(directory)},
+params={
+"directory": FileSystemUtilities.plainFileName(directory),
+"recursive": recursive,
+},
 callback=callback,
 )
 loop.exec()
 if not ok:
 @param ignore list of entries to be ignored (defaults to None)
 @type list of str (optional)
 @param recursive flag indicating a recursive search (defaults to True)
 @type bool (optional)
 @param dirsonly flag indicating to return only directories. When True it has
-precedence over the 'filesonly' parameter ((defaults to False)
+precedence over the 'filesonly' parameter (defaults to False)
 @type bool
 @return list of all files and directories in the tree rooted at path.
 The names are expanded to start with the given directory name.
 @rtype list of str
 @exception OSError raised in case the server reported an issue
 @param name name to be checked
 @type str
 @return flag indicating a directory
 @rtype bool
 """
-with contextlib.suppress(KeyError, OSError):
+try:
-result = self.stat(name, ["st_mode"])
+return stat.S_ISDIR(_RemoteFsCache[name]["mode"])
-return stat.S_ISDIR(result["st_mode"])
+except KeyError:
+with contextlib.suppress(KeyError, OSError):
+result = self.stat(name, ["st_mode"])
+return stat.S_ISDIR(result["st_mode"])
 return False
 def isfile(self, name):
 """
 @param name name to be checked
 @type str
 @return flag indicating a regular file
 @rtype bool
 """
-with contextlib.suppress(KeyError, OSError):
+try:
-result = self.stat(name, ["st_mode"])
+return stat.S_ISREG(_RemoteFsCache[name]["mode"])
-return stat.S_ISREG(result["st_mode"])
+except KeyError:
+with contextlib.suppress(KeyError, OSError):
+result = self.stat(name, ["st_mode"])
+return stat.S_ISREG(result["st_mode"])
 return False
 def exists(self, name):
 """
 nonlocal nameExists
 if reply == "Exists":
 nameExists = params["exists"]
 loop.quit()
+if name in _RemoteFsCache:
+return True
 if self.__serverInterface.isServerConnected():
 self.__serverInterface.sendJson(
 category=EricRequestCategory.FileSystem,
 request="Exists",
 params={"directory": FileSystemUtilities.plainFileName(directory)},
 callback=callback,
 )
 loop.exec()
+if ok:
+self.populateFsCache(directory)
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
 },
 callback=callback,
 )
 loop.exec()
+if ok:
+self.populateFsCache(directory)
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
 params={"directory": FileSystemUtilities.plainFileName(directory)},
 callback=callback,
 )
 loop.exec()
+if ok:
+self.removeFromFsCache(directory)
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
 },
 callback=callback,
 )
 loop.exec()
+if ok:
+with contextlib.suppress(KeyError):
+entry = _RemoteFsCache.pop(oldName)
+entry["path"] = newName
+entry["name"] = self.basename(newName)
+_RemoteFsCache[newName] = entry
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
 params={"filename": FileSystemUtilities.plainFileName(filename)},
 callback=callback,
 )
 loop.exec()
+if ok:
+with contextlib.suppress(KeyError):
+del _RemoteFsCache[filename]
 return ok, error
 else:
 return False, EricServerFileSystemInterface.NotConnectedMessage
 @param p path to be checked
 @type str
 @return flag indicating an absolute path
 @rtype bool
 """
-if self.__serverInterface.isServerConnected():
+if self.__serverPathSep == "\\":
-if self.__serverPathSep == "\\":
+s = FileSystemUtilities.plainFileName(p)[:3].replace("/", "\\")
-s = FileSystemUtilities.plainFileName(p)[:3].replace("/", "\\")
+return s.startswith("\\)") or s.startswith(":\\", 1)
-return s.startswith("\\)") or s.startswith(":\\", 1)
-else:
-return FileSystemUtilities.plainFileName(p).startswith("/")
 else:
-return os.path.isabs(p)
+return FileSystemUtilities.plainFileName(p).startswith("/")
 def abspath(self, p):
 """
 Public method to convert the given path to an absolute path.
 @param p path to be converted
 @type str
 @return absolute path
 @rtype str
 """
-if self.__serverInterface.isServerConnected():
+p = FileSystemUtilities.plainFileName(p)
-p = FileSystemUtilities.plainFileName(p)
+if not self.isabs(p):
-if not self.isabs(p):
+p = self.join(self.getcwd(), p)
-p = self.join(self.getcwd(), p)
+return FileSystemUtilities.remoteFileName(p)
-return FileSystemUtilities.remoteFileName(p)
-else:
-return os.path.abspath(p)
 def join(self, a, *p):
 """
 Public method to join two or more path name components using the path separator
 of the server side.
 @param *p list of additional path components
 @type list of str
 @return joined path name
 @rtype str
 """
-if self.__serverInterface.isServerConnected():
+path = a
-path = a
+for b in p:
-for b in p:
+if b.startswith(self.__serverPathSep):
-if b.startswith(self.__serverPathSep):
+path = b
-path = b
+elif not path or path.endswith(self.__serverPathSep):
-elif not path or path.endswith(self.__serverPathSep):
+path += b
-path += b
+else:
-else:
+path += self.__serverPathSep + b
-path += self.__serverPathSep + b
+return path
-return path
-else:
-return os.path.join(a, *p)
 def split(self, p):
 """
 Public method to split a path name.
 @type str
 @return tuple containing head and tail, where tail is everything after the last
 path separator.
 @rtype tuple of (str, str)
 """
-if self.__serverInterface.isServerConnected():
+if self.__serverPathSep == "\\":
-if self.__serverPathSep == "\\":
+# remote is a Windows system
-# remote is a Windows system
+normp = p.replace("/", "\\")
-normp = p.replace("/", "\\")
-else:
-# remote is a Posix system
-normp = p.replace("\\", "/")
-i = normp.rfind(self.__serverPathSep) + 1
-head, tail = normp[:i], normp[i:]
-if head and head != self.__serverPathSep * len(head):
-head = head.rstrip(self.__serverPathSep)
-return head, tail
 else:
-return os.path.split(p)
+# remote is a Posix system
+normp = p.replace("\\", "/")
+i = normp.rfind(self.__serverPathSep) + 1
+head, tail = normp[:i], normp[i:]
+if head and head != self.__serverPathSep * len(head):
+head = head.rstrip(self.__serverPathSep)
+return head, tail
 def splitext(self, p):
 """
 Public method to split a path name into a root part and an extension.
 @param p path name to be split
 @type str
 @return tuple containing the drive letter (incl. colon) and the path
 @rtype tuple of (str, str)
 """
-if self.__serverInterface.isServerConnected():
+plainp = FileSystemUtilities.plainFileName(p)
-plainp = FileSystemUtilities.plainFileName(p)
+if self.__serverPathSep == "\\":
-if self.__serverPathSep == "\\":
+# remote is a Windows system
-# remote is a Windows system
+normp = plainp.replace("/", "\\")
-normp = plainp.replace("/", "\\")
+if normp[1:2] == ":":
-if normp[1:2] == ":":
+return normp[:2], normp[2:]
-return normp[:2], normp[2:]
-else:
-return "", normp
 else:
-# remote is a Posix system
-normp = plainp.replace("\\", "/")
 return "", normp
 else:
-return os.path.splitdrive(p)
+# remote is a Posix system
+normp = plainp.replace("\\", "/")
+return "", normp
 def dirname(self, p):
 """
 Public method to extract the directory component of a path name.
 #######################################################################
 ## Methods for reading and writing files
 #######################################################################
-def readFile(self, filename, create=False):
+def readFile(self, filename, create=False, newline=None):
 """
 Public method to read a file from the eric-ide server.
 @param filename name of the file to read
 @type str
 @param create flag indicating to create an empty file, if it does not exist
 (defaults to False)
 @type bool (optional)
+@param newline determines how to parse newline characters from the stream
+(defaults to None)
+@type str (optional)
 @return bytes data read from the eric-ide server
 @rtype bytes
 @exception EricServerNotConnectedError raised to indicate a missing server
 connection
 @exception OSError raised in case the server reported an issue
 category=EricRequestCategory.FileSystem,
 request="ReadFile",
 params={
 "filename": FileSystemUtilities.plainFileName(filename),
 "create": create,
+"newline": "<<none>>" if newline is None else newline,
 },
 callback=callback,
 )
 loop.exec()
 if not ok:
 raise OSError(error)
 return bText
-def writeFile(self, filename, data, withBackup=False):
+def writeFile(self, filename, data, withBackup=False, newline=None):
 """
 Public method to write the data to a file on the eric-ide server.
 @param filename name of the file to write
 @type str
 @param data data to be written
-@type bytes
+@type bytes or QByteArray
 @param withBackup flag indicating to create a backup file first
 (defaults to False)
 @type bool (optional)
+@param newline determines how to parse newline characters from the stream
+(defaults to None)
+@type str (optional)
 @exception EricServerNotConnectedError raised to indicate a missing server
 connection
 @exception OSError raised in case the server reported an issue
 """
 loop = QEventLoop()
 if not self.__serverInterface.isServerConnected():
 raise EricServerNotConnectedError
 else:
+if isinstance(data, QByteArray):
+data = bytes(data)
 self.__serverInterface.sendJson(
 category=EricRequestCategory.FileSystem,
 request="WriteFile",
 params={
 "filename": FileSystemUtilities.plainFileName(filename),
 "filedata": str(base64.b85encode(data), encoding="ascii"),
 "with_backup": withBackup,
+"newline": "<<none>>" if newline is None else newline,
 },
 callback=callback,
 )
 loop.exec()
 },
 callback=callback,
 )
 loop.exec()
+if ok:
+self.removeFromFsCache(pathname)
 if not ok:
 raise OSError(error)
 #######################################################################
 ## Utility methods.
 cpath = f"{remoteMarker}{ellipsis}{tail}"
 if measure(cpath) <= width:
 return cpath
 tail = tail[1:]
 return ""
+#######################################################################
+## Remote file system cache methods.
+#######################################################################
+# TODO: change cache when file renamed/moved/deleted/...
+def populateFsCache(self, directory):
+"""
+Public method to populate the remote file system cache for a given directory.
+@param directory remote directory to be cached
+@type str
+@exception ValueError raised to indicate an empty directory
+"""
+if not directory:
+raise ValueError("The directory to be cached must not be empty.")
+try:
+listing = self.listdir(directory=directory, recursive=True)[2]
+for entry in listing:
+_RemoteFsCache[
+FileSystemUtilities.remoteFileName(entry["path"])
+] = entry
+print(f"Remote Cache Size: {len(_RemoteFsCache)} entries")
+except OSError as err:
+print("error in 'populateFsCache()':", str(err))
+def removeFromFsCache(self, directory):
+"""
+Public method to remove a given directory from the remote file system cache.
+@param directory remote directory to be removed
+@type str
+"""
+for entryPath in list(_RemoteFsCache.keys()):
+if entryPath.startswith(directory):
+del _RemoteFsCache[entryPath]
+print(f"Remote Cache Size: {len(_RemoteFsCache)} entries")

Mercurial Repositories > eric / file comparison

comparison: src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py

src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py