src/eric7/RemoteServer/EricServerFileSystemRequestHandler.py

branch
server
changeset 10539
4274f189ff78
child 10546
300487f5f517
diff -r 3308e8349e4c -r 4274f189ff78 src/eric7/RemoteServer/EricServerFileSystemRequestHandler.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/eric7/RemoteServer/EricServerFileSystemRequestHandler.py	Fri Feb 02 11:29:08 2024 +0100
@@ -0,0 +1,214 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
+#
+
+"""
+Module implementing the file system request handler of the eric-ide server.
+"""
+
+import os
+import stat
+import time
+
+from .EricRequestCategory import EricRequestCategory
+
+class EricServerFileSystemRequestHandler:
+    """
+    Class implementing the file system request handler of the eric-ide server.
+    """
+
+    def __init__(self, server):
+        """
+        Constructor
+
+        @param server reference to the eric-ide server object
+        @type EricServer
+        """
+        self.__server = server
+        
+        self.__requestMethodMapping = {
+            "Chdir": self.__chdir,
+            "Getcwd": self.__getcwd,
+            "Listdir": self.__listdir,
+            "Mkdir": self.__mkdir,
+            "Rmdir": self.__rmdir,
+            "Replace": self.__replace,
+            "Remove": self.__remove,
+        }
+
+    def handleRequest(self, request, params, reqestUuid):
+        """
+        Public method handling the received file system requests.
+
+        @param request request name
+        @type str
+        @param params dictionary containing the request parameters
+        @type dict
+        @param reqestUuid UUID of the associated request as sent by the eric IDE
+        @type str
+        """
+        try:
+            result = self.__requestMethodMapping[request](params)
+            self.__server.sendJson(
+                category=EricRequestCategory.FileSystem,
+                reply=request,
+                params=result,
+                reqestUuid=reqestUuid,
+            )
+
+        except KeyError:
+            self.__server.sendJson(
+                category=EricRequestCategory.FileSystem,
+                reply=request,
+                params={"Error": f"Request type '{request}' is not supported."},
+            )
+
+    def __chdir(self, params):
+        """
+        Private method to change the current working directory.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        directory = params["directory"]
+        try:
+            os.chdir(directory)
+            return {"ok": True}
+        except OSError as err:
+            return {
+                "ok": False,
+                "error": str(err),
+            }
+
+    def __getcwd(self, params):
+        """
+        Private method to report the current working directory.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        return {"directory": os.getcwd()}
+
+    def __listdir(self, params):
+        """
+        Private method to report a directory listing.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        directory = params["directory"]
+        if not directory:
+            directory = os.getcwd()
+
+        listing = []
+        for dirEntry in os.scandir(directory):
+            filestat = dirEntry.stat()
+            entry = {
+                "name": dirEntry.name,
+                "path": dirEntry.path,
+                "is_dir": dirEntry.is_dir(),
+                "is_file": dirEntry.is_file(),
+                "is_link": dirEntry.is_symlink(),
+                "mode": filestat.st_mode,
+                "mode_str": stat.filemode(filestat.st_mode),
+                "size": filestat.st_size,
+                "mtime": time.strftime(
+                    "%Y-%m-%d %H:%M:%S", time.localtime(filestat.st_mtime)
+                ),
+            }
+            listing.append(entry)
+
+        return {
+            "directory": directory,
+            "listing": listing,
+            "separator": os.sep,
+        }
+
+    def __mkdir(self, params):
+        """
+        Private method to create a new directory.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        directory = params["directory"]
+
+        try:
+            os.makedirs(directory)
+            return {"ok": True}
+        except OSError as err:
+            return {
+                "ok": False,
+                "error": str(err),
+            }
+
+    def __rmdir(self, params):
+        """
+        Private method to delete a directory.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        directory = params["directory"]
+
+        try:
+            os.rmdir(directory)
+            return {"ok": True}
+        except OSError as err:
+            return {
+                "ok": False,
+                "error": str(err),
+            }
+
+    def __replace(self, params):
+        """
+        Private method to replace (rename) a file or directory.
+
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        oldName = params["old_name"]
+        newName = params["new_name"]
+
+        try:
+            os.replace(oldName, newName)
+            return {"ok": True}
+        except OSError as err:
+            return {
+                "ok": False,
+                "error": str(err),
+            }
+
+    def __remove(self, params):
+        """
+        Private method to delete a directory.
+
+        @param params dictionary containing the request data
+        @type dict
+        @return dictionary containing the reply data
+        @rtype dict
+        """
+        filename = params["filename"]
+
+        try:
+            os.remove(filename)
+            return {"ok": True}
+        except OSError as err:
+            return {
+                "ok": False,
+                "error": str(err),
+            }

eric ide

mercurial