Main module of the library.

SFTP

Bases: SFTPClient

A wrapper around paramiko’s SFTPClient, with some improvements.

Source code in sftputil/sftputil.py
class SFTP(paramiko.sftp_client.SFTPClient):
    """A wrapper around paramiko’s SFTPClient, with some improvements."""

    def __init__(
        self,
        hostname,
        username,
        port=22,
        key_file=None,
        password=None,
        transport_kwargs={},
        channel_kwargs={},
    ):
        """Create a SFTP connection from scratch"""
        self.hostname = hostname
        # Create Socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        sock.connect((hostname, port))
        sock.settimeout(None)
        # Create Transport over Socket
        transport = paramiko.Transport(sock, **transport_kwargs)
        transport.start_client()
        # supposed to check for key in keys, but I don't much care right now
        # keys = paramiko.util.load_host_keys(os.path.expanduser("~/.ssh/known_hosts"))
        # key = transport.get_remote_server_key()
        if key_file is not None:
            pkey = paramiko.pkey.PKey.from_path(key_file)
            transport.auth_publickey(username, pkey)
        else:
            if password is not None:
                transport.auth_password(username, password, fallback=False)
            else:
                raise Exception("Must supply either key_file or password")
        # Create Channel over Transport over Socket
        channel = transport.open_session(**channel_kwargs)
        if channel is None:
            raise RuntimeError(
                "Cannot create a SSH channel toward '{}'".format(hostname)
            )
        channel.invoke_subsystem("sftp")
        # Create SFTP over Channel over Transport over Socket
        super().__init__(channel)

    def walk(self, remotepath):
        """Walk from a remote directory like os.walk

        Args:
            remotepath: (str|pathlib.Path) the top directory to walk from.
                If it points to a file, it yields nothing and ends immediately.
                If it points to nothing, it raises an error.

        Yields:
            Same tuple than `os.walk`: (path, folders, files)

        Raise:
            FileNotFoundError: if `remotepath` does not exist.
        """
        path = pathlib.Path(remotepath)
        if not self.exists(path):
            raise FileNotFoundError(path)
        if self.is_file(path):
            return
        files = []
        folders = []
        for fattr in self.listdir_iter(str(path)):
            if S_ISDIR(fattr.st_mode):
                folders.append(fattr.filename)
            else:
                files.append(fattr.filename)
        yield path, folders, files
        for folder in folders:
            new_path = path.joinpath(folder)
            for new_tuple in self.walk(new_path):
                yield new_tuple

    def sync_pull(
        self, remote_path, local_directory, recursive=True, ignore=None, keep_tree=False
    ):
        """Like rsync but with SFTP, to download files.

        Read `sftp.sync.sync_pull` documentation.
        """
        _sync_pull(self, remote_path, local_directory, recursive, ignore, keep_tree)

    def exists(self, path):
        """Whether the path points to an existing file or directory.

        Follow symlinks.

        Args:
            path: remote path to check

        Returns:
            (bool) True if the path exists, False else.
        """
        try:
            self.stat(str(path))
        except FileNotFoundError:
            return False
        return True

    def is_dir(self, path):
        """Whether the path points to an existing directory.

        Follow symlinks.

        Args:
            path: remote path to check

        Returns:
            (bool) True if the path points to a dir, False else.
        """
        try:
            attr = self.stat(str(path))
        except FileNotFoundError:
            return False
        if S_ISDIR(attr.st_mode):
            return True
        return False

    def is_file(self, path):
        """Whether the path points to an existing file.

        Follow symlinks.

        Args:
            path: remote path to check

        Returns:
            (bool) True if the path points to a file, False else.
        """
        try:
            attr = self.stat(str(path))
        except FileNotFoundError:
            return False
        if S_ISREG(attr.st_mode):
            return True
        return False

    def iglob(self, pathname, recursive=True):
        """SFTP implementation of glob.iglob, but lighter"""
        yield from _iglob(self, pathname, recursive, False)

    def glob(self, pathname, recursive=True):
        """SFTP implementation of glob.glob, but lighter"""
        return list(self.iglob(pathname, recursive))

__init__(hostname, username, port=22, key_file=None, password=None, transport_kwargs={}, channel_kwargs={})

Create a SFTP connection from scratch

Source code in sftputil/sftputil.py
def __init__(
    self,
    hostname,
    username,
    port=22,
    key_file=None,
    password=None,
    transport_kwargs={},
    channel_kwargs={},
):
    """Create a SFTP connection from scratch"""
    self.hostname = hostname
    # Create Socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(3)
    sock.connect((hostname, port))
    sock.settimeout(None)
    # Create Transport over Socket
    transport = paramiko.Transport(sock, **transport_kwargs)
    transport.start_client()
    # supposed to check for key in keys, but I don't much care right now
    # keys = paramiko.util.load_host_keys(os.path.expanduser("~/.ssh/known_hosts"))
    # key = transport.get_remote_server_key()
    if key_file is not None:
        pkey = paramiko.pkey.PKey.from_path(key_file)
        transport.auth_publickey(username, pkey)
    else:
        if password is not None:
            transport.auth_password(username, password, fallback=False)
        else:
            raise Exception("Must supply either key_file or password")
    # Create Channel over Transport over Socket
    channel = transport.open_session(**channel_kwargs)
    if channel is None:
        raise RuntimeError(
            "Cannot create a SSH channel toward '{}'".format(hostname)
        )
    channel.invoke_subsystem("sftp")
    # Create SFTP over Channel over Transport over Socket
    super().__init__(channel)

exists(path)

Whether the path points to an existing file or directory.

Follow symlinks.

Parameters:
  • path

    remote path to check

Returns:
  • (bool) True if the path exists, False else.

Source code in sftputil/sftputil.py
def exists(self, path):
    """Whether the path points to an existing file or directory.

    Follow symlinks.

    Args:
        path: remote path to check

    Returns:
        (bool) True if the path exists, False else.
    """
    try:
        self.stat(str(path))
    except FileNotFoundError:
        return False
    return True

glob(pathname, recursive=True)

SFTP implementation of glob.glob, but lighter

Source code in sftputil/sftputil.py
def glob(self, pathname, recursive=True):
    """SFTP implementation of glob.glob, but lighter"""
    return list(self.iglob(pathname, recursive))

iglob(pathname, recursive=True)

SFTP implementation of glob.iglob, but lighter

Source code in sftputil/sftputil.py
def iglob(self, pathname, recursive=True):
    """SFTP implementation of glob.iglob, but lighter"""
    yield from _iglob(self, pathname, recursive, False)

is_dir(path)

Whether the path points to an existing directory.

Follow symlinks.

Parameters:
  • path

    remote path to check

Returns:
  • (bool) True if the path points to a dir, False else.

Source code in sftputil/sftputil.py
def is_dir(self, path):
    """Whether the path points to an existing directory.

    Follow symlinks.

    Args:
        path: remote path to check

    Returns:
        (bool) True if the path points to a dir, False else.
    """
    try:
        attr = self.stat(str(path))
    except FileNotFoundError:
        return False
    if S_ISDIR(attr.st_mode):
        return True
    return False

is_file(path)

Whether the path points to an existing file.

Follow symlinks.

Parameters:
  • path

    remote path to check

Returns:
  • (bool) True if the path points to a file, False else.

Source code in sftputil/sftputil.py
def is_file(self, path):
    """Whether the path points to an existing file.

    Follow symlinks.

    Args:
        path: remote path to check

    Returns:
        (bool) True if the path points to a file, False else.
    """
    try:
        attr = self.stat(str(path))
    except FileNotFoundError:
        return False
    if S_ISREG(attr.st_mode):
        return True
    return False

sync_pull(remote_path, local_directory, recursive=True, ignore=None, keep_tree=False)

Like rsync but with SFTP, to download files.

Read sftp.sync.sync_pull documentation.

Source code in sftputil/sftputil.py
def sync_pull(
    self, remote_path, local_directory, recursive=True, ignore=None, keep_tree=False
):
    """Like rsync but with SFTP, to download files.

    Read `sftp.sync.sync_pull` documentation.
    """
    _sync_pull(self, remote_path, local_directory, recursive, ignore, keep_tree)

walk(remotepath)

Walk from a remote directory like os.walk

Parameters:
  • remotepath

    (str|pathlib.Path) the top directory to walk from. If it points to a file, it yields nothing and ends immediately. If it points to nothing, it raises an error.

Yields:
  • Same tuple than os.walk: (path, folders, files)

Raise

FileNotFoundError: if remotepath does not exist.

Source code in sftputil/sftputil.py
def walk(self, remotepath):
    """Walk from a remote directory like os.walk

    Args:
        remotepath: (str|pathlib.Path) the top directory to walk from.
            If it points to a file, it yields nothing and ends immediately.
            If it points to nothing, it raises an error.

    Yields:
        Same tuple than `os.walk`: (path, folders, files)

    Raise:
        FileNotFoundError: if `remotepath` does not exist.
    """
    path = pathlib.Path(remotepath)
    if not self.exists(path):
        raise FileNotFoundError(path)
    if self.is_file(path):
        return
    files = []
    folders = []
    for fattr in self.listdir_iter(str(path)):
        if S_ISDIR(fattr.st_mode):
            folders.append(fattr.filename)
        else:
            files.append(fattr.filename)
    yield path, folders, files
    for folder in folders:
        new_path = path.joinpath(folder)
        for new_tuple in self.walk(new_path):
            yield new_tuple