The sftp: scheme

Needs uripath: pip install "sftppath[uri]".

sftppath.uri

The sftp: scheme of uripath: SFTPURIPath, a URI path that delegates to SFTPPath, and SFTPURIBackend, the options its connections are made with.

Importing this module needs uripath: pip install "sftppath[uri]". uripath loads it by itself, through the uripath.schemes entry point, the first time a path of the scheme is used.

SFTPURIBackend(backend_class=None, **options)

Bases: object

How the connections of sftp: paths are made: the backend class and the options of its constructor (connect_opts, ssh_config, timeout, ...). backend_for(source) builds the backend for a URI's server and account, once for as long as a path or a connection holds it, so paths to one server share their connections.

backend_class is a BaseSFTPBackend subclass; without one, the class SFTPPATH_BACKEND selects is used. The user, password and port of the URI are given to it; options are given as they are.

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

Source code in src/sftppath/uri.py

def __init__(
    self,
    backend_class: "type[BaseSFTPBackend] | None" = None,
    **options,
) -> None:
    self.backend_class = backend_class
    self.options = options
    # The backends made, by folded endpoint, while a path or a cached
    # connection holds one.
    self._backends: "_weakref.WeakValueDictionary[Source, BaseSFTPBackend]" = (
        _weakref.WeakValueDictionary()
    )
    self._lock = _threading.Lock()

backend_class = backend_class instance-attribute

options = options instance-attribute

backend_for(source)

The backend for the server and account source names.

Source code in src/sftppath/uri.py

def backend_for(self, source: Source) -> BaseSFTPBackend:
    """The backend for the server and account `source` names."""
    key = source._replace(host=fold_host(source.host))
    with self._lock:
        found = self._backends.get(key)
        if found is None:
            cls = self.backend_class or _resolve_default_backend_cls()
            user, password = source.parsed_userinfo()
            found = self._backends[key] = cls(
                str(source.host),
                source.port,
                user=user or None,
                password=password or None,
                **self.options,
            )
        return found

close()

Closes every connection the paths of this backend hold, whichever thread opened it, and returns when they are closed. Harmless when called again; a later use connects afresh.

Source code in src/sftppath/uri.py

def close(self) -> None:
    """Closes every connection the paths of this backend hold, whichever
    thread opened it, and returns when they are closed. Harmless when
    called again; a later use connects afresh."""
    with self._lock:
        held = list(self._backends.values())
    for backend in held:
        backend.close()

SFTPURIPath

Bases: DelegatingURIPath

sftp: scheme: full read/write access, done by an SFTPPath on the connections of backend (an SFTPURIBackend, or a BaseSFTPBackend used as it is for every path whatever host the URI names). See "backend selection" in SFTPPath; one of its SSH-library extras is needed.

ssh_config -- default ~/.ssh/config, None for none, a path or an iterable of paths -- is carried to every path derived from this one on the same endpoint and dropped, with the backend, when the derived path is on another host.

backend instance-attribute

checksum(algorithm='md5')

Source code in src/sftppath/uri.py

def checksum(self, algorithm: str = "md5") -> str:
    with self._delegated():
        return self._core_path.checksum(algorithm)

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

Copy with the backend's concurrent fan-out: a directory copied to a path of the same server and tree is done by the native walk of SFTPPath.copy(); every other copy is Path.copy(). progress is not called during the native fan-out.

Source code in src/sftppath/uri.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 the backend's concurrent fan-out: a directory copied to a
    path of the same server and tree is done by the native walk of
    `SFTPPath.copy()`; every other copy is `Path.copy()`. `progress` is not
    called during the native fan-out."""
    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 not None:

        def asked(error, path):
            return offering(error, self._from_core(path))

        extra = {"on_error": asked}
    if (
        not recursive
        or not self._core_path.backend.supports_tree
        # The fan-out writes every destination file over THIS path's connection: only a
        # target on the same host and tree (two supplied backends may reach different
        # servers) may use it.
        or not isinstance(target, SFTPURIPath)
        # The native walk keeps the setuid, setgid and sticky bits only between
        # two paths of one class, and both are an `SFTPPath` here.
        or type(target) is not type(self)
        or not self._same_location(target)
        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,
            **({} if offering is None else {"on_error": offering}),
        )
    moved = self._core_of(target)
    with self._delegated((str(moved), str(target))):
        return self._core_path._copy_tree(
            moved,
            overwrite=overwrite,
            follow_symlinks=follow_symlinks,
            preserve_metadata=preserve_metadata,
            ignore_error=ignore_error,
            **extra,
        )

hardlink_to(target)

Source code in src/sftppath/uri.py

def hardlink_to(self, target: "SFTPURIPath | URI | str"):
    if isinstance(target, URI) and not self._same_location(target):
        raise NotImplementedError(
            f"hardlink_to() cannot link across hosts: {self} -> {target}"
        )
    target_path = target.path if isinstance(target, URI) else str(target)
    with self._delegated():
        self._core_path.hardlink_to(target_path)

readlink()

Source code in src/sftppath/uri.py

def readlink(self) -> "SFTPURIPath":
    # The raw target string, unresolved and verbatim (no dot-segment folding), as
    # pathlib's readlink(): it is a result, not a destination. A relative target has
    # no host; with this path's authority it would have no printable URI.
    with self._delegated():
        target = self._core_path._readlink_text()
    source = self.source if target.startswith("/") else _NO_SOURCE
    return self._from_parsed_parts(source, target, None, None)

rename(target)

Source code in src/sftppath/uri.py

def rename(self, target: "SFTPURIPath | URI | str"):
    # A plain str target is a sibling rename (relative to self's parent): a
    # literal name, not re-parsed as a URI. An existing target is replaced via
    # `posix-rename@openssh.com` where the server has it.
    target = self._rename_target(target)
    destination = self.with_path(target.path)
    moved = self._core_of(target)
    with self._delegated((str(moved), str(destination))):
        self._core_path.rename(moved)
    # 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 it is the replace() of SFTPPath; onto anything else it is Path.replace(), which move(overwrite=True) uses too.

Source code in src/sftppath/uri.py

def replace(self, target: "SFTPURIPath | URI | str"):
    """Rename this entry to `target`, replacing what is there, and return
    `target` as a path. Onto a path of this server it is the `replace()` of
    `SFTPPath`; onto anything else it is `Path.replace()`, which
    `move(overwrite=True)` uses too."""
    if isinstance(target, str):
        target = self._coerce_target(target)
    if type(target) is type(self) and self._same_location(target):
        moved = self._core_of(target)
        with self._delegated((str(moved), str(target))):
            self._core_path.replace(moved)
        return 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 asked this method
        # again: the target is not on this server, so `move()` copies.
        raise OSError(_errno.EXDEV, "Cross-device link", str(target.path))
    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/uri.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.
    if (
        recursive
        and follow_symlinks is False
        and self._core_path.backend.supports_tree
    ):
        on_error = ignore_error
        if callable(ignore_error):

            def on_error(error, path):
                return ignore_error(error, self._from_core(path))

        with self._delegated():
            return self._core_path.rm(
                recursive=True,
                missing_ok=missing_ok,
                ignore_error=on_error,
                follow_symlinks=follow_symlinks,
                follow_binds=follow_binds,
            )
    return super().rm(
        recursive=recursive,
        missing_ok=missing_ok,
        ignore_error=ignore_error,
        follow_symlinks=follow_symlinks,
        follow_binds=follow_binds,
    )

stat(*, follow_symlinks=True)

Source code in src/sftppath/uri.py

def stat(self, *, follow_symlinks=True):
    hint = self._pop_stat_hint()
    if hint is not None and (
        not follow_symlinks
        # The hint comes from listdir_attr(), which never resolves
        # symlinks: it is the answer to an lstat, and for an entry that
        # says what it is and is not a link that is the answer to a stat.
        or (_stat.S_IFMT(hint.st_mode) and not hint.is_symlink())
    ):
        return hint
    with self._delegated():
        return self._core_path.stat(follow_symlinks=follow_symlinks)

supported_checksums()

Source code in src/sftppath/uri.py

def supported_checksums(self) -> "_ty.FrozenSet[str]":
    with self._delegated():
        return self._core_path.supported_checksums()

with_source(source)

Source code in src/sftppath/uri.py

def with_source(self, source):
    inst = super().with_source(source)
    if isinstance(inst, SFTPURIPath):
        inst._ssh_config = self._config_for(source)
    return inst