I've written a function to copy files remote-to-remote using SCP command, paramiko, pexpect, and paramiko-sftp.

I want to make my code efficient; I'm new to Python and I don't know how.

Here's my logic in 3 steps:

  1. Find checksum of a file by executing cksum command in remote-1.
  2. Executing SCP command from my local to copy file remote-1 to remote-2.
  3. Again do checksum and compare both to give status.

class remote_file_operations():
    Class which supports remote file operations
    def __init__(self, host=None, username=None, password=None, port=22):
        *Method description :* Initialize and setup connection

        :param host: Remote Host
        :type host: str
        :param username: Remote Host Username
        :type username: str
        :param password: Remote Host password
        :type password: str
        :param port: port to connect, default port is 22
        :type port: int
        if host and username and password:
            self.sftp = None
            self.sftp_open = False
            self.client = paramiko.client.SSHClient()
                self.client.connect(host, port=port,
                                username=username, password=password, timeout=10)
            except paramiko.ssh_exception.AuthenticationException:
                sys.exit("[Authentication Error] Please check user: '%s' or password" % username)
            except TimeoutError:
                sys.exit("Connection Refused")
            except socket.gaierror:
                sys.exit("Host Name: '%s' not found" % host)

    def _open_sftp_connection(self):
        *Method description :* Opens an SFTP connection if not already open
        if not self.sftp_open:
            self.sftp = self.client.open_sftp()
            self.sftp_open = True
        return self.sftp_open

    def file_check_sum(user, host, password, file_paths):
        *Method description :* Give check sum value by cksum command in remote.

        :param user: username :type user: str
        :param host: hostname :type host: str
        :param password: password :type password: str
        :param file_paths: list of file paths :type file_paths: list
        :return check_sum: dict with key as filename and value as checksum :type check_sum: dict
        status = True
        logger.console("Finding checksum ...")
        check_sum = dict()
        rmt_ses = remote_file_operations(host, user, password)
        for file_path in file_paths:
            _, stdout, stderr = rmt_ses.client.exec_command("ls %s" % file_path.replace("\\", ""))
            if b"No such file or directory" not in stderr.read():
                for abs_file_path in stdout:
                    cmd = "cksum %s" % abs_file_path.replace("\n", "").replace("\r", "")
                    (_, stdout, stderr) = rmt_ses.client.exec_command(cmd)
                    sum_line = stdout.read().decode("utf-8").replace(abs_file_path, "")
                    file_name = abs_file_path.split("/")[-1]
                    check_sum[file_name] = sum_line.replace("\n", "").strip()
                logger.error("File doesn't exist: %s:%s" % (host, file_path))
                status = False
        return status, check_sum

    def scp_remote_remote(host_1, user_1, password_1, host_2, user_2, password_2, src, dest):
        #pylint: disable=too-many-arguments,try-except-raise,too-many-locals,too-many-branches
        *Method description :* SCP files from one remote to another remote

        :param host_1: Remote host-1  :type host_1: str
        :param user_1: Remote user-1  :type user_1: str
        :param password_1: Remote Host-1 password  :type password_1: str
        :param host_2: Remote host-2  :type host_2: str
        :param user_2: Remote user-2  :type user_2: str
        :param password_2: Remote Host-2 password  :type password_2: str
        :param src: Source dir/file  :type src: str
        :param dest: Destination dir/file  :type dest: str
        :return output: Output generated while scp
        :type output: list
        exp_pwd_2, status, output = None, False, list()
        check, src_check_sum = remote_file_operations.file_check_sum(user_1, host_1, password_1, [src])
        if check:
            command = "scp -r %s@%s:%s %s@%s:%s" % (user_1, host_1, src, user_2, host_2, dest)
            logger.console("Executing: %s" % command)
                child = pexpect.spawn(command, timeout=COPY_TIMEOUT)
                logger.console("Please wait while copying files from %s to %s ..." % (host_1, host_2))
                exp_pwd_1 = child.expect(["password:", "want to continue connecting"], timeout=10)
                if exp_pwd_1 == 0:
                elif exp_pwd_1 == 1:
                exp_pwd_2 = child.expect(["password:", "want to continue connecting"], timeout=10)
                if exp_pwd_2 == 0:
                elif exp_pwd_2 == 1:
                for line in child.before.decode("utf-8").split("\r\n"):
                    for sub_line in line.replace("\r", " ").split("ETA"):
                    output.append(line.replace("\r", " "))
                msg = "scp: %s: " % dest
                if msg + "Permission denied" in output or msg + "No such file or directory" in output:
                    logger.error("Please check the dest path: '%s' exists or have access" % dest)
                elif "scp: /ome/virtuora: No such file or directory" % dest in output:
                    logger.error("Please check the dest path: '%s' exists or have access" % dest)
                    status = True
            except pexpect.exceptions.TIMEOUT:
                if "Permission denied" in child.before.decode("utf-8") or "password:"\
                                         in child.before.decode("utf-8"):
                    logger.error("Please check the username or password")
            except pexpect.exceptions.EOF:
                if "ld not resolve hostname" in child.before.decode("utf-8"):
                    logger.error("Please check the Hostnames: '%s', '%s'" % (host_1, host_2))
            if status:
                dst_file = [dest + "/" + src.split("/")[-1]]
                status, dst_check_sum = remote_file_operations.file_check_sum(user_2, host_2, password_2, dst_file)
                if status:
                    for src_file, src_check in src_check_sum.items():
                        for dst_file, dst_check in dst_check_sum.items():
                            if src_file == dst_file and src_check == dst_check:
                                logger.console("%s Transfered Successfully" % (src_file))
                                status = True
                                logger.console("Something went wrong in file transferring")
        return status, output

1 Answer 1


Type hints

Move your type documentation:

    :type host: str

to actual type hints:

def __init__(self, host: Optional[str]=None, username: Optional[str]=None, password: Optional[str]=None, port: int=22):

String interpolation

            sys.exit("Host Name: '%s' not found" % host)

can be

            sys.exit(f"Host Name: '{host}' not found")

Object closure

    rmt_ses = remote_file_operations(host, user, password)
    # ...

So a few things. remote_file_operations should be RemoteFileOperations since it's a class. Also, it's not clear to me that this file_check_sum method should be a static on the class - it makes just as much sense for it to be a function in global scope.

Also, reaching into the object to client.close() is poor coupling; the class should be turned into a context manager whose __exit__ calls its client.close(). Then, file_check_sum would use RemoteFileOperations in a with.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.