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), + }