Mercurial Repositories > eric / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the file system interface to the eric-ide server.
"""

import contextlib

from PyQt6.QtCore import QEventLoop, QObject

from eric7.RemoteServer.EricRequestCategory import EricRequestCategory


class EricServerFileSystemInterface(QObject):
    """
    Class implementing the file system interface to the eric-ide server.
    """

    def __init__(self, serverInterface):
        """
        Constructor

        @param serverInterface reference to the eric-ide server interface
        @type EricServerInterface
        """
        super().__init__(parent=serverInterface)

        self.__serverInterface = serverInterface

    def getcwd(self):
        """
        Public method to get the current working directory of the eric-ide server.

        @return current working directory of the eric-ide server
        @rtype str
        """
        loop = QEventLoop()
        cwd = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal cwd

            if reply == "Getcwd":
                cwd = params["directory"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Getcwd",
            params={},
            callback=callback,
        )

        loop.exec()
        return cwd

    def chdir(self, directory):
        """
        Public method to change the current working directory of the eric-ide server.

        @param directory absolute path of the working directory to change to
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Chdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Chdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def listdir(self, directory=""):
        """
        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)
        @return tuple containing the listed directory, the path separartor and the
            directory listing. Each directory listing entry contains a dictionary
            with the relevant data.
        @rtype tuple of (str, str, dict)
        """
        if directory is None:
            # sanitize the directory in case it is None
            directory = ""

        loop = QEventLoop()
        listedDirectory = ""
        separator = ""
        listing = []

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal listedDirectory, listing, separator

            if reply == "Listdir":
                listedDirectory = params["directory"]
                listing = params["listing"]
                separator = params["separator"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Listdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return listedDirectory, separator, listing

    def mkdir(self, directory):
        """
        Public method to create a new directory on the eric-ide server.

        @param directory absolute path of the new directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Mkdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Mkdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def rmdir(self, directory):
        """
        Public method to delete a directory on the eric-ide server.

        @param directory absolute path of the directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Rmdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Rmdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def replace(self, oldName, newName):
        """
        Public method to rename a file or directory.

        @param oldName current name of the file or directory
        @type str
        @param newName new name for the file or directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Replace":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Replace",
            params={"old_name": oldName, "new_name": newName},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def remove(self, filename):
        """
        Public method to delete a file on the eric-ide server.

        @param filename absolute path of the file
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Remove":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Remove",
            params={"filename": filename},
            callback=callback,
        )

        loop.exec()
        return ok, error

    #######################################################################
    ## Methods for reading and writing files
    #######################################################################

    def readFile(self, filename):
        """
        Public method to read a file from the eric-ide server.

        @param filename name of the file to read
        @type str
        @return tuple containing an OK flag, the read data and an error string in
            case of an issue
        @rtype tuple of (bool, str, str)
        """
        # TODO: 'readFile()' not implemented yet

    def writeFile(self, filename, data):
        """
        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 str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        # TODO: 'writeFile()' not implemented yet