sftp:

Needs the sftp extra (paramiko backend) or the sftp-async extra (asyncssh backend).

pathlib_next.uri.schemes.sftp

BaseSftpBackend

Bases: object

Protocol for obtaining a paramiko-shaped SFTPClient for a Source. Subclass this to plug in custom connection handling (e.g. tests mock it directly, no real server); SftpBackend (paramiko, _paramiko.py) and AsyncsshSftpBackend (_asyncssh.py, optional extra) are the real implementations. Connection caching is each backend's own responsibility -- client() is expected to return an already-cached-or-freshly-opened, ready-to-use client; SftpPath itself does no per-backend branching anywhere.

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 protocols/checksum.py::NativeChecksum.checksum for why this is a hard contract, not a style choice). Both real backends implement it through _checkfile.py::CheckFileSftpBackend, each supplying only the extended request itself.

Source code in src/pathlib_next/uri/schemes/sftp/__init__.py

@_utils.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 `protocols/checksum.py::NativeChecksum.checksum` for why this
    is a hard contract, not a style choice). Both real backends
    implement it through `_checkfile.py::CheckFileSftpBackend`, each
    supplying only the extended request itself.
    """
    ...

close()

Close every connection this backend has cached. A no-op here; both real backends override it. The backend stays usable: the next client() call reconnects.

Source code in src/pathlib_next/uri/schemes/sftp/__init__.py

def close(self) -> None:
    """Close every connection this backend has cached. A no-op here;
    both real backends override it. The backend stays usable: the next
    `client()` call reconnects."""

supported_checksums(path)

Advisory set of algorithm names checksum() can currently produce against path's server connection (see 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.py::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/pathlib_next/uri/schemes/sftp/__init__.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
    `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.py::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()

SftpPath(*uris, **options)

Bases: UriPath

sftp: scheme: full read/write access, auto-selecting between a paramiko (sync) and an asyncssh (async, bridged) backend -- see "backend selection" above. Requires the sftp extra (paramiko) or sftp-async extra (asyncssh). Also implements 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.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if not uri:
            uri = ""
        if isinstance(uri, Uri):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(path).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

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/pathlib_next/uri/schemes/sftp/__init__.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

copy(target, *, overwrite=False, follow_symlinks=True, preserve_metadata=True, recursive=False, ignore_error=None, progress=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 fallback path below, but not called during the concurrent native fan-out itself -- see docs/divergences.md's "Deliberate extensions" section for the documented limitation.

Source code in src/pathlib_next/uri/schemes/sftp/__init__.py

def copy(
    self,
    target,
    *,
    overwrite=False,
    follow_symlinks=True,
    preserve_metadata=True,
    recursive=False,
    ignore_error=None,
    progress=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 fallback path below, but **not** called during
    the concurrent native fan-out itself -- see `docs/divergences.md`'s
    "Deliberate extensions" section for the documented limitation.
    """
    try:
        from ._asyncssh import AsyncsshSftpBackend, _concurrent_copy, _run
    except ImportError:
        # paramiko-only install ('sftp' extra): generic copy.
        AsyncsshSftpBackend = None

    if (
        AsyncsshSftpBackend is None
        or not isinstance(self.backend, AsyncsshSftpBackend)
        or not recursive
        # The fan-out writes every destination file over THIS path's
        # connection: only a target on the same host may use it.
        or not isinstance(target, SftpPath)
        or not self._same_location(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,
        )

    if _contains(self, target):
        # As the generic copy(): checked before anything is created, or
        # the new directory is listed and copied into itself.
        raise OSError(
            _errno.EINVAL, "Cannot copy a directory into itself", str(target)
        )

    if target.exists():
        if not target.is_dir():
            raise FileExistsError(
                _errno.EEXIST, _os.strerror(_errno.EEXIST), str(target)
            )
        if not overwrite:
            raise FileExistsError(
                _errno.EEXIST, _os.strerror(_errno.EEXIST), str(target)
            )
    else:
        target.mkdir()

    coro = _concurrent_copy(
        self,
        target,
        overwrite=overwrite,
        follow_symlinks=follow_symlinks,
        preserve_metadata=preserve_metadata,
        max_concurrency=self.backend.max_concurrency,
        ignore_error=ignore_error,
        # Resolved on this thread, never on the bridge loop (see rm()).
        aclient=self._sftpclient._aclient,
    )
    return _run(coro, None)

supported_checksums()

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

Source code in src/pathlib_next/uri/schemes/sftp/__init__.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.py::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)

pathlib_next.uri.schemes.sftp._paramiko.SftpBackend(connect_opts=None, hostkeypolicy=None, ssh_config=_DEFAULT_SSH_CONFIG, *, known_hosts=_DEFAULT_KNOWN_HOSTS, timeout=DEFAULT_TIMEOUT)

Bases: CheckFileSftpBackend

Connects via paramiko.SSHClient using connect_opts merged with the Source's host/port/userinfo. client() caches per (self, source, calling-thread) -- see _CACHED_CLIENTS above -- and replaces (and closes) a cached client whose connection has dropped; close() closes every connection the backend opened.

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 paramiko.BadHostKeyException. Opt-out, in code only: SftpBackend(connect_opts, paramiko.AutoAddPolicy(), known_hosts=None) accepts any server key -- a network man-in-the-middle then receives the URI 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/pathlib_next/uri/schemes/sftp/_paramiko.py

def __init__(
    self,
    connect_opts=None,
    hostkeypolicy=None,
    ssh_config=_DEFAULT_SSH_CONFIG,
    *,
    known_hosts=_DEFAULT_KNOWN_HOSTS,
    timeout: "float | None" = DEFAULT_TIMEOUT,
) -> None:
    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

close()

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

Source code in src/pathlib_next/uri/schemes/sftp/_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)

pathlib_next.uri.schemes.sftp._asyncssh.AsyncsshSftpBackend(connect_opts=None, *, max_concurrency=None, sftp_version=4, ssh_config=_DEFAULT_SSH_CONFIG, timeout=_DEFAULT_TIMEOUT)

Bases: CheckFileSftpBackend

sftp: backend using asyncssh instead of paramiko. Selected automatically when asyncssh is importable (see backend selection in sftp/__init__.py), or explicitly via backend=AsyncsshSftpBackend() /PATHLIB_NEXT_SFTP_BACKEND=asyncssh. Connections are cached per (self, source) (see _ConnectionCache above) and served through a single shared background asyncio loop (see _run above) -- 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.

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 fails the connection. Opt-out, in code only: AsyncsshSftpBackend(connect_opts={"known_hosts": None}) accepts any server key -- a network man-in-the-middle then receives the URI password. Any other asyncssh known_hosts value (a file, a list of keys) is passed through as given.

timeout (default _DEFAULT_TIMEOUT, 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/pathlib_next/uri/schemes/sftp/_asyncssh.py

def __init__(
    self,
    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" = _DEFAULT_TIMEOUT,
):
    if max_concurrency is None:
        max_concurrency = self.DEFAULT_MAX_CONCURRENCY
    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

close()

Close every cached connection this backend opened.

Source code in src/pathlib_next/uri/schemes/sftp/_asyncssh.py

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