SFTP paths

sftppath

pathlib-style paths for SFTP servers, on paramiko or asyncssh.

SFTPPath needs no URI and imports neither SSH library until a backend is built: SFTPBackend (paramiko) and AsyncSSHBackend (asyncssh) are resolved on first use (PEP 562), so import sftppath needs neither extra and never imports uripath; naming one whose library is missing raises an ImportError that names the extra. The sftp: scheme of uripath is sftppath.uri.

DEFAULT_PORT = 22 module-attribute

The port a connection uses when neither the caller nor the ssh_config of the host names one.

DEFAULT_SSH_CONFIG = _Sentinel(__name__, 'DEFAULT_SSH_CONFIG') module-attribute

SFTPPath(*segments, backend=None, host=None, port=None, user=None, password=None, ssh_config=DEFAULT_SSH_CONFIG, timeout=UNSET, connect_opts=None, **options)

Bases: TreePath

A path on an SFTP server, over paramiko or asyncssh (see "backend selection" in _select). A request needs the paramiko or the asyncssh extra; naming, joining and comparing paths needs neither. Also implements pathapi.protocols.checksum.NativeChecksum (checksum(), delegating to self.backend.checksum()) -- native on both backends via the filexfer draft's check-file-handle extension where the server implements it (OpenSSH does not), NotImplementedError (falls back to streaming) on a server without that extension.

SFTPPath("/srv/x", host="h", user="u") builds the path's own backend from the endpoint keywords and the connect options (ssh_config=, timeout=, connect_opts=, and the keywords of the backend class that SFTPPATH_BACKEND selects); backend= instead shares one connection between paths built separately, or plugs in a BaseSFTPBackend of your own. Giving backend= together with any of those keywords is a TypeError, and so is giving neither (unless a segment is an SFTPPath, whose backend is kept). Nothing connects when a path is built. Every path derived from one (/, joinpath, parent, with_name, iterdir, relative_to) keeps its backend. Names are read from the server root: srv/x and /srv/x are one file.

Source code in src/sftppath/_path.py

def __init__(
    self,
    *segments: "str | SFTPPath",
    backend: "BaseSFTPBackend | None" = None,
    host: "str | None" = None,
    port: "int | None" = None,
    user: "str | None" = None,
    password: "str | None" = None,
    ssh_config: object = DEFAULT_SSH_CONFIG,
    timeout: "float | None | object" = UNSET,
    connect_opts: "dict | None" = None,
    **options,
):
    text, donor = self._join(segments)
    endpoint = (
        host is not None
        or port is not None
        or user is not None
        or password is not None
        or ssh_config is not DEFAULT_SSH_CONFIG
        or timeout is not UNSET
        or connect_opts is not None
        or bool(options)
    )
    if backend is not None and endpoint:
        raise TypeError(
            "pass either backend= or the endpoint keywords "
            "(host=, port=, user=, ...), not both"
        )
    if backend is None:
        if endpoint:
            if host is None:
                raise TypeError(
                    "host= is required with the other endpoint keywords"
                )
            backend = self._new_backend(
                host,
                port,
                user,
                password,
                ssh_config,
                timeout,
                connect_opts,
                options,
            )
        elif donor is not None:
            backend = donor._backend
        else:
            raise TypeError("an SFTPPath needs host= or backend=")
    self._segments = self._parse(text)
    self._backend = backend

backend property

The backend (SFTPBackend, AsyncSSHBackend or a custom BaseSFTPBackend) holding the connections of this path.

as_uri()

Source code in src/sftppath/_path.py

def as_uri(self) -> str:
    backend = self._backend
    host = getattr(backend, "host", None)
    if host is None:
        raise NotImplementedError("as_uri() needs a backend with a host")
    netloc = f"[{host}]" if ":" in host else host
    if backend.user is not None:
        netloc = f"{_urlparse.quote(backend.user, safe='')}@{netloc}"
    if backend.port is not None:
        netloc = f"{netloc}:{backend.port}"
    return f"sftp://{netloc}{_urlparse.quote(self._wirepath, safe='/')}"

checksum(algorithm='md5')

protocols.checksum.NativeChecksum implementation: delegates to self.backend.checksum() (the check-file-handle SFTP extension on both backends -- see BaseSFTPBackend.checksum). Any failure that isn't already NotImplementedError (a server that doesn't advertise the extension, an unsupported algorithm, a transport-level error) is also translated to NotImplementedError: this method's whole contract is "raise if a genuine digest can't be produced", and a caller (e.g. PathSyncer) must be able to fall back to streaming on ANY such failure, not just the backend's own explicit signal.

Source code in src/sftppath/_path.py

def checksum(self, algorithm: str = "md5") -> str:
    """`protocols.checksum.NativeChecksum` implementation: delegates to
    `self.backend.checksum()` (the `check-file-handle` SFTP extension
    on both backends -- see `BaseSFTPBackend.checksum`). Any failure
    that isn't already
    `NotImplementedError` (a server that doesn't advertise the
    extension, an unsupported algorithm, a transport-level error) is
    also translated to `NotImplementedError`: this method's whole
    contract is "raise if a genuine digest can't be produced", and a
    caller (e.g. `PathSyncer`) must be able to fall back to streaming
    on ANY such failure, not just the backend's own explicit signal.
    """
    try:
        return self._backend.checksum(self, algorithm)
    except NotImplementedError:
        raise
    except Exception as error:
        raise NotImplementedError(
            f"native checksum unavailable: {error}"
        ) from error

chmod(mode, *, follow_symlinks=True)

Source code in src/sftppath/_path.py

def chmod(self, mode: int | str, *, follow_symlinks: bool = True):
    mode = _providers.parse_mode(mode)
    if follow_symlinks:
        return self._sftpclient.chmod(self._wirepath, mode)
    if not self._backend.supports_lchmod:
        raise NotImplementedError("chmod(follow_symlinks=False)")
    return self._sftpclient.chmod(self._wirepath, mode, follow_symlinks=False)

copy(target, *, overwrite=False, follow_symlinks=True, preserve_metadata=True, recursive=False, ignore_error=None, progress=None, on_error=None)

Copy with concurrent fan-out on the asyncssh backend.

When using the asyncssh backend with recursive=True on a directory, child copies are fanned out as concurrent native requests, bounded by backend.max_concurrency (requests in flight, and files open at once); the whole copy has no wall-clock timeout. progress is honored on the generic single-file path below, but not called during the concurrent native fan-out itself. on_error(error, path) is Path.copy()'s predicate; the native fan-out calls it from a worker thread, once per failing child.

Source code in src/sftppath/_path.py

def copy(
    self,
    target,
    *,
    overwrite=False,
    follow_symlinks=True,
    preserve_metadata=True,
    recursive=False,
    ignore_error=None,
    progress=None,
    on_error=None,
):
    """Copy with concurrent fan-out on the asyncssh backend.

    When using the asyncssh backend with `recursive=True` on a
    directory, child copies are fanned out as concurrent native requests,
    bounded by `backend.max_concurrency` (requests in flight, and files
    open at once); the whole copy has no wall-clock timeout. `progress`
    is honored on the generic single-file path below, but not called
    during the concurrent native fan-out itself.
    `on_error(error, path)` is `Path.copy()`'s predicate; the native
    fan-out calls it from a worker thread, once per failing child.
    """
    offering = copy_error_predicate(ignore_error, on_error)
    # Passed on only when given: a backend or a `copy()` override without
    # the keyword has no parameter for it.
    extra = {} if offering is None else {"on_error": offering}
    if (
        not recursive
        or not self._backend.supports_tree
        # The fan-out writes every destination file over THIS path's connection: only a
        # target on the same server and tree (two supplied backends may reach different
        # servers) may use it.
        or not isinstance(target, SFTPPath)
        or not self._same_filesystem(target)
        # A link copied as a link is the generic copy's job.
        or (not follow_symlinks and self.is_symlink())
        or not self.is_dir()
    ):
        return super().copy(
            target,
            overwrite=overwrite,
            follow_symlinks=follow_symlinks,
            preserve_metadata=preserve_metadata,
            recursive=recursive,
            ignore_error=ignore_error,
            progress=progress,
            **extra,
        )

    return self._copy_tree(
        target,
        overwrite=overwrite,
        follow_symlinks=follow_symlinks,
        preserve_metadata=preserve_metadata,
        ignore_error=ignore_error,
        **extra,
    )

hardlink_to(target)

Source code in src/sftppath/_path.py

def hardlink_to(self, target: "SFTPPath | str"):
    if not self._backend.supports_hardlink:
        raise NotImplementedError("hardlink_to() requires the asyncssh backend")
    if isinstance(target, SFTPPath):
        if not self._same_filesystem(target):
            raise NotImplementedError(
                f"hardlink_to() cannot link across servers: {self} -> {target}"
            )
        target_path = target._wirepath
    else:
        target_path = str(target)
    self._sftpclient.link(target_path, self._wirepath)

is_local()

Whether the server is this machine, as far as its name says: a loopback or unspecified address, or localhost. A host name is not resolved, and a backend with no host counts as local. PathSyncer skips its quick_check for two local sides.

Source code in src/sftppath/_path.py

def is_local(self) -> bool:
    """Whether the server is this machine, as far as its name says: a
    loopback or unspecified address, or `localhost`. A host name is not
    resolved, and a backend with no host counts as local. `PathSyncer`
    skips its `quick_check` for two local sides."""
    host = getattr(self._backend, "host", None)
    if host is None:
        return True
    name = str(host).strip("[]").split("%", 1)[0].lower()
    if name == "localhost" or name.endswith(".localhost"):
        return True
    try:
        address = _ipaddress.ip_address(name)
    except ValueError:
        return False
    return address.is_loopback or address.is_unspecified

iterdir()

Source code in src/sftppath/_path.py

def iterdir(self):
    for name, _stat in self._scandir():
        yield self.with_segments(*self.segments, name)

readlink()

Source code in src/sftppath/_path.py

def readlink(self) -> "SFTPPath":
    # The target is a result, not a destination: a path on this backend.
    return self._rebuild(self._readlink_text())

rename(target)

Source code in src/sftppath/_path.py

def rename(self, target: "SFTPPath | str"):
    # A plain str target is a literal sibling name. An existing target is
    # replaced via `posix-rename@openssh.com`; plain v3 RENAME refuses
    # (FileExistsError).
    if isinstance(target, str):
        target = self.parent / target
    elif not isinstance(target, SFTPPath) or not self._same_filesystem(target):
        raise NotImplementedError("rename() onto another server")
    destination = self._rebuild(target.as_posix())
    self._rename_on_wire(destination)
    # pathlib returns the new path.
    return destination

replace(target)

Rename this entry to target, replacing what is there, and return target as a path.

Onto a path of this server the request is posix-rename@openssh.com where the server has it: one request, atomic, and if it fails both entries are as they were. A server that lacks the extension, or that refuses it without saying why, gets what Path.replace() does: the file at target is removed and then this entry is renamed. Onto anything else this is Path.replace(), which move(overwrite=True) uses too.

Source code in src/sftppath/_path.py

def replace(self, target: "SFTPPath | str"):
    """Rename this entry to `target`, replacing what is there, and return
    `target` as a path.

    Onto a path of this server the request is `posix-rename@openssh.com`
    where the server has it: one request, atomic, and if it fails both
    entries are as they were. A server that lacks the extension, or that
    refuses it without saying why, gets what `Path.replace()` does: the
    file at `target` is removed and then this entry is renamed. Onto
    anything else this is `Path.replace()`, which `move(overwrite=True)`
    uses too."""
    if isinstance(target, str):
        target = self.parent / target
    pending = _REPLACING.get()
    if pending is not None and pending[0] is self and pending[1] is target:
        # `Path.replace()` below reached `move()`, which found a file at
        # `target` and asked this method to swap it in.
        if not self._same_filesystem(target):
            raise OSError(_errno.EXDEV, "Cross-device link", target._wirepath)
        target.unlink(missing_ok=True)
        return self.rename(target)
    if type(target) is type(self) and self._same_filesystem(target):
        client = self._sftpclient
        posix_rename = getattr(client, "posix_rename", None)
        if posix_rename is not None and not _posix_rename_known_unsupported(client):
            try:
                posix_rename(self._wirepath, target._wirepath)
            except NotImplementedError:
                _mark_posix_rename_unsupported(client)
            except OSError as error:
                # A refusal that names its cause is the answer, unless a directory stands at
                # `target`: `Path.replace()` checks those before it removes anything.
                if error.errno is not None and not self._is_directory_at(target):
                    raise
            else:
                return target
    token = _REPLACING.set((self, target))
    try:
        return super().replace(target)
    finally:
        _REPLACING.reset(token)

rm(recursive=False, missing_ok=False, ignore_error=False, *, follow_symlinks=False, follow_binds=False)

Source code in src/sftppath/_path.py

def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: bool | _ty.Callable[[Exception, _ty.Self], bool] = False,
    *,
    follow_symlinks=False,
    follow_binds=False,
):
    check_follow("follow_symlinks", follow_symlinks)
    check_follow("follow_binds", follow_binds)
    # The native walker unlinks a symlink and never follows one, which is
    # the default policy; any other `follow_symlinks` takes the generic
    # walk. A server has no bindings, so `follow_binds` decides nothing.
    if (
        not recursive
        or follow_symlinks is not False
        or not self._backend.supports_tree
    ):
        return super().rm(
            recursive=recursive,
            missing_ok=missing_ok,
            ignore_error=ignore_error,
            follow_symlinks=follow_symlinks,
            follow_binds=follow_binds,
        )

    on_error = None
    if ignore_error:
        on_error = (
            ignore_error
            if callable(ignore_error)
            else lambda _err, _path: bool(ignore_error)
        )

    return self._backend.tree_rm(self, missing_ok=missing_ok, on_error=on_error)

rmdir()

Source code in src/sftppath/_path.py

def rmdir(self):
    try:
        return self._sftpclient.rmdir(self._wirepath)
    except OSError as error:
        translated = self._directory_error(error, check_empty=True)
        if translated is None:
            raise
        raise translated from error

stat(*, follow_symlinks=True)

Source code in src/sftppath/_path.py

def stat(self, *, follow_symlinks=True):
    if follow_symlinks:
        return self._sftpclient.stat(self._wirepath)
    else:
        return self._sftpclient.lstat(self._wirepath)

supported_checksums()

protocols.checksum.NativeChecksum implementation: delegates to self.backend.supported_checksums(self) -- a real per-connection probe on both backends (see _checkfile.CheckFileSFTPBackend.supported_checksums), so this is empty when the connected server doesn't implement check-file-handle (OpenSSH never does).

Source code in src/sftppath/_path.py

def supported_checksums(self) -> "_ty.FrozenSet[str]":
    """`protocols.checksum.NativeChecksum` implementation: delegates to
    `self.backend.supported_checksums(self)` -- a real per-connection
    probe on both backends (see
    `_checkfile.CheckFileSFTPBackend.supported_checksums`), so this
    is empty when the connected server doesn't implement
    `check-file-handle` (OpenSSH never does).
    """
    return self._backend.supported_checksums(self)

unlink(missing_ok=False)

Source code in src/sftppath/_path.py

def unlink(self, missing_ok=False):
    # No exists() pre-check: it follows symlinks, so a dangling link
    # read as missing and was never removed (breaking
    # symlink_to(force=True)). pathlib ignores only ENOENT from the call.
    try:
        return self._sftpclient.remove(self._wirepath)
    except FileNotFoundError:
        if not missing_ok:
            raise
    except OSError as error:
        translated = self._file_error(error)
        if translated is None:
            raise
        raise translated from error

SFTPEndpoint

Bases: NamedTuple

The server and account a backend connects to: what two backends must share for their paths to name the same files. The password is not part of it, and the host name is in lower case. port is None when the caller named none (the ssh_config Port of the host, else 22, applies).

host instance-attribute

port instance-attribute

user instance-attribute

BaseSFTPBackend

Bases: object

Contract for obtaining a ready-to-use, paramiko-shaped SFTPClient for one server. Subclass this to plug in custom connection handling (tests mock it directly, no real server); SFTPBackend (paramiko) and AsyncSSHBackend (asyncssh) are the real implementations. Connection caching is each backend's own responsibility: client() is expected to return an already-cached-or-freshly-opened client, and SFTPPath does no per-backend branching anywhere.

endpoint, when a subclass sets it, is a hashable value naming the server and account: two backends with equal endpoints serve the same files.

A context manager: leaving the block calls close().

endpoint = None class-attribute instance-attribute

supports_hardlink = False class-attribute instance-attribute

supports_lchmod = False class-attribute instance-attribute

supports_tree = False class-attribute instance-attribute

checksum(path, algorithm)

Backend-native digest for path's content, e.g. via the filexfer draft's check-file-handle SFTP extension (implemented by some servers, e.g. ProFTPD's mod_sftp; OpenSSH has no such extension). Raises NotImplementedError (the base/default here) when the backend has no such capability at all, and MUST also raise it -- not return a value -- when the server doesn't advertise algorithm specifically (see pathapi.protocols.checksum.NativeChecksum.checksum for why this is a hard contract, not a style choice). Both real backends implement it through _checkfile.CheckFileSFTPBackend, each supplying only the extended request itself.

Source code in src/sftppath/_backend.py

@_notimplemented
def checksum(self, path: "SFTPPath", algorithm: str) -> str:
    """Backend-native digest for `path`'s content, e.g. via the
    filexfer draft's `check-file-handle` SFTP extension (implemented by
    some servers, e.g. ProFTPD's mod_sftp; OpenSSH has no such
    extension). Raises
    `NotImplementedError` (the base/default here) when the backend has
    no such capability at all, and MUST also raise it -- not return a
    value -- when the server doesn't advertise `algorithm` specifically
    (see `pathapi.protocols.checksum.NativeChecksum.checksum` for
    why this is a hard contract, not a style choice). Both real
    backends implement it through
    `_checkfile.CheckFileSFTPBackend`, each supplying only the extended
    request itself.
    """
    ...

client()

A ready-to-use SFTP client for this backend's server, cached or newly opened.

Source code in src/sftppath/_backend.py

@_notimplemented
def client(self):
    """A ready-to-use SFTP client for this backend's server, cached or
    newly opened."""

close()

Closes every connection this backend has cached, whichever thread opened it, and returns when they are closed. A no-op here; both real backends override it. Harmless when called again; the backend stays usable, and the next client() call reconnects.

Source code in src/sftppath/_backend.py

def close(self) -> None:
    """Closes every connection this backend has cached, whichever thread
    opened it, and returns when they are closed. A no-op here; both real
    backends override it. Harmless when called again; the backend stays
    usable, and the next `client()` call reconnects."""

supported_checksums(path)

Advisory set of algorithm names checksum() can currently produce against path's server connection (see pathapi.protocols.checksum.NativeChecksum.supported_checksums). Empty here (the default): no native-hashing support at all. Both real backends override this with a per-connection probe against path (neither library exposes the server's extension list -- see _checkfile.CheckFileSFTPBackend.supported_checksums). Takes a path argument (unlike a bare capability flag) because the only reliable way to know is to actually try the extension against a real file.

Source code in src/sftppath/_backend.py

def supported_checksums(self, path: "SFTPPath") -> "_ty.FrozenSet[str]":
    """Advisory set of algorithm names `checksum()` can currently
    produce against `path`'s server connection (see
    `pathapi.protocols.checksum.NativeChecksum.supported_checksums`).
    Empty here (the default): no native-hashing support at all. Both real
    backends override this with a per-connection probe against `path`
    (neither library exposes the server's extension list -- see
    `_checkfile.CheckFileSFTPBackend.supported_checksums`). Takes a
    `path` argument (unlike a bare capability flag) because the only
    reliable way to know is to actually try the extension against a
    real file.
    """
    return frozenset()

tree_copy(path, target, *, overwrite, follow_symlinks, preserve_metadata, ignore_error, on_error=None)

Copy the directory path into the existing target directory, on the server both are on. Only called when supports_tree. on_error(error, path), when given, decides whether a failure of one child is skipped; ignore_error is then None.

Source code in src/sftppath/_backend.py

def tree_copy(
    self,
    path: "SFTPPath",
    target: "SFTPPath",
    *,
    overwrite: bool,
    follow_symlinks: bool,
    preserve_metadata: bool,
    ignore_error,
    on_error=None,
) -> None:
    """Copy the directory `path` into the existing `target` directory,
    on the server both are on. Only called when `supports_tree`.
    `on_error(error, path)`, when given, decides whether a failure of one
    child is skipped; `ignore_error` is then None."""
    raise NotImplementedError("tree_copy()")

tree_rm(path, *, missing_ok, on_error)

Remove path and everything below it, never following a link. on_error(error, path) decides whether a failure is skipped. Only called when supports_tree.

Source code in src/sftppath/_backend.py

def tree_rm(self, path: "SFTPPath", *, missing_ok: bool, on_error) -> None:
    """Remove `path` and everything below it, never following a link.
    `on_error(error, path)` decides whether a failure is skipped. Only
    called when `supports_tree`."""
    raise NotImplementedError("tree_rm()")

CheckFileSFTPBackend

Bases: BaseSFTPBackend

A BaseSFTPBackend whose checksum()/supported_checksums() speak check-file-handle. A subclass supplies _check_file_request(); the handle is opened (and always closed) through its client's paramiko- shaped open().

checksum(path, algorithm)

Server-side digest of path through check-file-handle. A server that answers "operation unsupported" for algorithm is not asked for it again on this connection (an OpenSSH server costs one request per algorithm, not one per file); other algorithms are still asked for. Other failures propagate as raised, and refuse nothing; SFTPPath.checksum() translates them to NotImplementedError.

Source code in src/sftppath/_checkfile.py

def checksum(self, path: "SFTPPath", algorithm: str) -> str:
    """Server-side digest of `path` through `check-file-handle`. A server
    that answers "operation unsupported" for `algorithm` is not asked for
    it again on this connection (an OpenSSH server costs one request per
    algorithm, not one per file); other algorithms are still asked for.
    Other failures propagate as raised, and refuse nothing;
    `SFTPPath.checksum()` translates them to `NotImplementedError`."""
    if algorithm not in DIGEST_SIZES:
        raise NotImplementedError(
            f"{EXTENSION}: {algorithm!r} is not an algorithm the draft names"
        )
    client = self.client()
    record = _support(client)
    if algorithm in record.refused:
        raise NotImplementedError(
            f"{EXTENSION}: the server does not support {algorithm}"
        )
    # The extension hashes an *open handle*, not a bare path -- open
    # read-only and unbuffered (it is never read), and always close, so
    # an attempt never leaks a handle whether it succeeds or not.
    file = client.open(path._wirepath, "r", 0)
    try:
        try:
            payload = self._check_file_request(client, file, algorithm)
        except Exception as error:
            if refused(error):
                record.refused.add(algorithm)
            raise
    finally:
        file.close()
    digest = parse_reply(payload, algorithm)
    record.produced.add(algorithm)
    return digest

supported_checksums(path)

Per-connection probe: try checksum() against path once and remember the answer for this connected client. path must already exist and be readable, or the probe's own open() fails for an unrelated reason (a missing file), which is reported as empty here but not remembered -- this method is advisory, never raises.

Only the FIRST algorithm is actually attempted: a server either implements the extension (and then the draft's algorithm set) or it doesn't. Once any digest has been produced the answer is the draft's set less the algorithms the server refused; before that it is empty, and checksum() still asks for any algorithm not refused.

Source code in src/sftppath/_checkfile.py

def supported_checksums(self, path: "SFTPPath") -> "frozenset[str]":
    """Per-connection probe: try `checksum()` against `path` once and
    remember the answer for this connected client. `path` must already
    exist and be readable, or the probe's own `open()` fails for an
    unrelated reason (a missing file), which is reported as empty here
    but not remembered -- this method is advisory, never raises.

    Only the FIRST algorithm is actually attempted: a server either
    implements the extension (and then the draft's algorithm set) or it
    doesn't. Once any digest has been produced the answer is the draft's
    set less the algorithms the server refused; before that it is empty,
    and `checksum()` still asks for any algorithm not refused.
    """
    client = self.client()
    record = _support(client)
    if not record.produced and not record.probed:
        if ALGORITHMS[0] not in record.refused:
            try:
                self.checksum(path, ALGORITHMS[0])
            except NotImplementedError:
                pass
            except Exception:
                # Any other failure (missing file, transport hiccup, ...)
                # isn't evidence one way or the other about extension
                # support -- don't remember an inconclusive probe.
                return frozenset()
        record.probed = True
    if record.produced:
        return frozenset(a for a in ALGORITHMS if a not in record.refused)
    return frozenset()

sftppath.SFTPBackend(host, port=None, *, user=None, password=None, connect_opts=None, hostkeypolicy=None, ssh_config=DEFAULT_SSH_CONFIG, known_hosts=_DEFAULT_KNOWN_HOSTS, timeout=DEFAULT_TIMEOUT)

Bases: CheckFileSFTPBackend

The connections to one SFTP server and account, on paramiko.SSHClient: connect_opts merged with host, port, user and password. Nothing connects when the backend is built. client() caches per (self, calling thread) -- see _CACHED_CLIENTS above -- and replaces (and closes) a cached client whose connection has dropped; close() closes every connection the backend opened, on any thread, and is harmless when called again.

host may be a name from the user's ssh_config: HostName, Port, User and IdentityFile of the matching block apply where the caller named no port, user or key. Host keys are verified by default. known_hosts (default: the user's ~/.ssh/known_hosts plus any ssh_config UserKnownHostsFile for the host; None loads no file; a path or iterable of paths loads exactly those) is loaded read-only, and hostkeypolicy (default paramiko.RejectPolicy()) decides what happens to a key found in none of them. A key that differs from a known one always raises SFTPHostKeyError, as one the policy refuses does. Opt-out, in code only: SFTPBackend(host, hostkeypolicy=paramiko.AutoAddPolicy(), known_hosts=None) accepts any server key -- a network man-in-the-middle then receives the password.

timeout (default 30 s) is the default for paramiko's timeout, banner_timeout, auth_timeout and channel_timeout connect options; a value in connect_opts wins, and timeout=None leaves them unset. Individual SFTP requests on an established connection are not bounded.

ssh_config support is paramiko's own (HostName, Port, User, IdentityFile, ProxyCommand) plus Include. ProxyJump is not supported: a host whose config sets it raises NotImplementedError unless connect_opts supplies a sock -- use the asyncssh backend, or an equivalent ProxyCommand.

Source code in src/sftppath/_paramiko.py

def __init__(
    self,
    host: str,
    port: "int | None" = None,
    *,
    user: "str | None" = None,
    password: "str | None" = None,
    connect_opts=None,
    hostkeypolicy=None,
    ssh_config=DEFAULT_SSH_CONFIG,
    known_hosts=_DEFAULT_KNOWN_HOSTS,
    timeout: "float | None" = DEFAULT_TIMEOUT,
) -> None:
    if not host:
        raise ValueError("an SFTP backend needs a host")
    self.host = str(host)
    self.port = None if port is None else int(port)
    self.user = user
    self.password = password
    self.connect_opts = {} if connect_opts is None else connect_opts
    self.hostkeypolicy = (
        _paramiko.RejectPolicy() if hostkeypolicy is None else hostkeypolicy
    )
    self.ssh_config = ssh_config
    self.known_hosts = known_hosts
    self.timeout = timeout

DEFAULT_TIMEOUT = 30.0 class-attribute instance-attribute

__slots__ = ('host', 'port', 'user', 'password', 'connect_opts', 'hostkeypolicy', 'ssh_config', 'known_hosts', 'timeout') class-attribute instance-attribute

connect_opts = {} if connect_opts is None else connect_opts instance-attribute

endpoint property

host = str(host) instance-attribute

hostkeypolicy = _paramiko.RejectPolicy() if hostkeypolicy is None else hostkeypolicy instance-attribute

known_hosts = known_hosts instance-attribute

password = password instance-attribute

port = None if port is None else int(port) instance-attribute

ssh_config = ssh_config instance-attribute

timeout = timeout instance-attribute

user = user instance-attribute

__reduce__()

Source code in src/sftppath/_paramiko.py

def __reduce__(self):
    return (
        _rebuild_backend,
        (
            type(self),
            {
                "host": self.host,
                "port": self.port,
                "user": self.user,
                "password": self.password,
                "connect_opts": self.connect_opts,
                "hostkeypolicy": self.hostkeypolicy,
                "ssh_config": self.ssh_config,
                "known_hosts": self.known_hosts,
                "timeout": self.timeout,
            },
        ),
    )

__repr__()

Source code in src/sftppath/_paramiko.py

def __repr__(self) -> str:
    shown = [repr(self.host)]
    if self.port is not None:
        shown.append(str(self.port))
    if self.user is not None:
        shown.append(f"user={self.user!r}")
    return f"{type(self).__name__}({', '.join(shown)})"

client()

Source code in src/sftppath/_paramiko.py

def client(self):
    thread_id = _thread.get_ident()
    if (self, thread_id) not in _CACHED_CLIENTS.cache:
        _close_clients_of_ended_threads()
    client = _CACHED_CLIENTS(self, thread_id)
    if not _client_is_alive(client):
        client = _CACHED_CLIENTS.invalidate(self, thread_id)
    return client

close()

Close every cached connection this backend opened, on any thread.

Source code in src/sftppath/_paramiko.py

def close(self) -> None:
    """Close every cached connection this backend opened, on any thread."""
    for key in list(_CACHED_CLIENTS.cache):
        if key[0] is self:
            _CACHED_CLIENTS.discard(*key)

opts()

The keyword arguments SSHClient.connect() is called with: the connect_opts merged with the ssh_config of the host (a ProxyCommand starts its process here).

Source code in src/sftppath/_paramiko.py

def opts(self):
    """The keyword arguments `SSHClient.connect()` is called with: the
    `connect_opts` merged with the ssh_config of the host (a
    `ProxyCommand` starts its process here)."""
    _check_host(self.host)
    host = self.host
    config = _lookup_ssh_config(host, self.ssh_config)
    connect_ops = {
        **self.connect_opts,
        "hostname": config.get("hostname", host),
        "port": self.port or int(config.get("port", DEFAULT_PORT)),
    }
    if self.timeout is not None:
        for key in ("timeout", "banner_timeout", "auth_timeout", "channel_timeout"):
            connect_ops.setdefault(key, self.timeout)
    if self.user:
        connect_ops["username"] = self.user
    elif "username" not in connect_ops and "user" in config:
        connect_ops["username"] = str(config["user"])
    if self.password:
        connect_ops["password"] = self.password
    if "key_filename" not in connect_ops and "identityfile" in config:
        connect_ops["key_filename"] = list(config["identityfile"])
    if "sock" not in connect_ops:
        if config.get("proxycommand"):
            connect_ops["sock"] = _paramiko.ProxyCommand(
                _expand_proxy_command(
                    str(config["proxycommand"]),
                    host=host,
                    hostname=str(connect_ops["hostname"]),
                    port=connect_ops["port"],
                    user=connect_ops.get("username"),
                )
            )
        elif str(config.get("proxyjump") or "none").lower() != "none":
            # Connecting directly would silently bypass the jump host.
            raise NotImplementedError(
                f"ssh_config ProxyJump ({config['proxyjump']}) for host "
                f"{self.host!r} is not supported by the paramiko SFTP "
                "backend -- use the asyncssh backend ('asyncssh' extra), "
                "an equivalent ProxyCommand, or connect_opts['sock']"
            )
    return connect_ops

transport()

Connects and returns the paramiko.Transport.

Source code in src/sftppath/_paramiko.py

def transport(self) -> _paramiko.Transport:
    """Connects and returns the `paramiko.Transport`."""
    opts = self.opts()
    client = _paramiko.SSHClient()
    try:
        for path in self._known_hosts_files():
            # load_system_host_keys(): read-only, never written back --
            # an AutoAddPolicy must not edit the user's files either.
            client.load_system_host_keys(path)
        client.set_missing_host_key_policy(self.hostkeypolicy)
        client.connect(**opts)
        transport = client.get_transport()
        if not transport:
            raise _NoTransport("connect() produced no transport")
    except BaseException as error:
        # A failed connect/auth still leaves a running Transport thread
        # and an open socket (or a ProxyCommand subprocess) behind.
        client.close()
        sock = opts.get("sock")
        if sock is not None and "sock" not in self.connect_opts:
            try:
                sock.close()
            except Exception:
                pass
        translated = _connect_error(error, self.endpoint)
        if translated is None:
            raise
        raise translated from None
    return transport

sftppath.AsyncSSHBackend(host, port=None, *, user=None, password=None, connect_opts=None, max_concurrency=None, sftp_version=4, ssh_config=DEFAULT_SSH_CONFIG, timeout=_bridge._DEFAULT_TIMEOUT)

Bases: CheckFileSFTPBackend

The connections to one SFTP server and account, on asyncssh instead of paramiko. Selected automatically when asyncssh is importable (see backend selection in SFTPPath), or explicitly via backend=AsyncSSHBackend(host) / SFTPPATH_BACKEND=asyncssh. Nothing connects when the backend is built. The connection is cached per backend (see _ConnectionCache) and served through a single shared background asyncio loop (see _run) -- not fork-safe (a fork()ed child inherits a dead loop thread; detected via stored PID, loop+cache lazily recreated when os.getpid() changes). close() closes every connection the backend opened, returns when they are closed, and is harmless when called again.

host may be a name from the user's ssh_config: asyncssh applies the matching block where the caller named no port, user or key. Host keys are verified by default: asyncssh checks the server key against ~/.ssh/known_hosts and the ssh_config's UserKnownHostsFile, and an unknown or changed key raises SFTPHostKeyError. Opt-out, in code only: AsyncSSHBackend(host, connect_opts={"known_hosts": None}) accepts any server key -- a network man-in-the-middle then receives the password. Any other asyncssh known_hosts value (a file, a list of keys) is passed through as given.

timeout (default 60 s) bounds a single request -- connect, stat, open, mkdir, rename, ... -- and a timed-out request is cancelled and raises TimeoutError. Recursive copy()/rm(), streaming file reads/writes (read_bytes(), write_bytes(), chunked copies) and a native checksum() (the server hashes the whole file) have no wall-clock bound; asyncssh's own connect_timeout, login_timeout and keepalive_interval connect options detect a dead peer there. timeout=None disables the per-request bound too.

Source code in src/sftppath/_asyncssh/_backend.py

def __init__(
    self,
    host: str,
    port: "int | None" = None,
    *,
    user: "str | None" = None,
    password: "str | None" = None,
    connect_opts: "dict[str, _ty.Any] | None" = None,
    max_concurrency: "int | None" = None,
    sftp_version: int = 4,
    ssh_config=DEFAULT_SSH_CONFIG,
    timeout: "float | None" = _bridge._DEFAULT_TIMEOUT,
):
    if not host:
        raise ValueError("an SFTP backend needs a host")
    if max_concurrency is None:
        max_concurrency = self.DEFAULT_MAX_CONCURRENCY
    self.host = str(host)
    self.port = None if port is None else int(port)
    self.user = user
    self.password = password
    self.connect_opts = {} if connect_opts is None else dict(connect_opts)
    if "config" not in self.connect_opts:
        if ssh_config is None:
            self.connect_opts["config"] = None
        elif ssh_config is not DEFAULT_SSH_CONFIG:
            self.connect_opts["config"] = ssh_config
    self.max_concurrency = max_concurrency
    self.sftp_version = sftp_version
    self.timeout = timeout

DEFAULT_MAX_CONCURRENCY = 16 class-attribute instance-attribute

__slots__ = ('host', 'port', 'user', 'password', 'connect_opts', 'max_concurrency', 'sftp_version', 'timeout') class-attribute instance-attribute

connect_opts = {} if connect_opts is None else dict(connect_opts) instance-attribute

endpoint property

host = str(host) instance-attribute

max_concurrency = max_concurrency instance-attribute

password = password instance-attribute

port = None if port is None else int(port) instance-attribute

sftp_version = sftp_version instance-attribute

supports_hardlink = True class-attribute instance-attribute

supports_lchmod = True class-attribute instance-attribute

supports_tree = True class-attribute instance-attribute

timeout = timeout instance-attribute

user = user instance-attribute

__reduce__()

Source code in src/sftppath/_asyncssh/_backend.py

def __reduce__(self):
    # `connect_opts` already holds the ssh_config as `config`.
    return (
        _rebuild_backend,
        (
            type(self),
            {
                "host": self.host,
                "port": self.port,
                "user": self.user,
                "password": self.password,
                "connect_opts": self.connect_opts,
                "max_concurrency": self.max_concurrency,
                "sftp_version": self.sftp_version,
                "timeout": self.timeout,
            },
        ),
    )

__repr__()

Source code in src/sftppath/_asyncssh/_backend.py

def __repr__(self) -> str:
    shown = [repr(self.host)]
    if self.port is not None:
        shown.append(str(self.port))
    if self.user is not None:
        shown.append(f"user={self.user!r}")
    return f"{type(self).__name__}({', '.join(shown)})"

client()

Source code in src/sftppath/_asyncssh/_backend.py

def client(self) -> _SyncSFTPClient:
    try:
        entry = _CACHE.get_or_create(
            (self,),
            lambda: _bridge._run(
                _aconnect(
                    self.host,
                    self.port,
                    self.user,
                    self.password,
                    connect_opts=self.connect_opts,
                    sftp_version=self.sftp_version,
                    timeout=self.timeout,
                ),
                self.timeout,
            ),
        )
    except _asyncssh.Error as error:
        raise _transport_error(error, self.endpoint) from None
    return entry.client

close()

Close every cached connection this backend opened.

Source code in src/sftppath/_asyncssh/_backend.py

def close(self) -> None:
    """Close every cached connection this backend opened."""
    _CACHE.close_backend(self)

tree_copy(path, target, *, overwrite, follow_symlinks, preserve_metadata, ignore_error, on_error=None)

Source code in src/sftppath/_asyncssh/_backend.py

def tree_copy(
    self,
    path,
    target,
    *,
    overwrite,
    follow_symlinks,
    preserve_metadata,
    ignore_error,
    on_error=None,
):
    return _bridge._run(
        _concurrent_copy(
            path,
            target,
            overwrite=overwrite,
            follow_symlinks=follow_symlinks,
            preserve_metadata=preserve_metadata,
            max_concurrency=self.max_concurrency,
            ignore_error=ignore_error,
            on_error=on_error,
            aclient=path._sftpclient._aclient,
        ),
        None,
    )

tree_rm(path, *, missing_ok, on_error)

Source code in src/sftppath/_asyncssh/_backend.py

def tree_rm(self, path, *, missing_ok, on_error):
    return _bridge._run(
        _concurrent_rm(
            path,
            max_concurrency=self.max_concurrency,
            missing_ok=missing_ok,
            on_error=on_error,
            aclient=path._sftpclient._aclient,
        ),
        None,
    )