ftp: and ftps:

uripath.schemes.FTPURIPath(*uris, **options)

Bases: DelegatingURIPath

ftp:/ftps: scheme: full read/write access via the stdlib FTP client, done by a pathapi.ftp.FTPPath on the connections of backend (a thread-keyed LRU cache; a thread's connections close when it ends). A listing prefers MLSD (RFC 3659 -- type/size/modify facts in one round trip) and a stat MLST of the entry itself where FEAT lists it, else its parent's MLSD; servers that don't support them fall back to NLST for listing (names only) and SIZE/CWD for stat (no portable "not found vs. is a directory" distinction in that path).

Source code in src/uripath/_uri.py

def __init__(self, *uris: URILike, **options):
    if options:
        self._check_options(options)
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | URI] = []
    for uri in uris:
        if uri is None or (isinstance(uri, (str, bytes)) and 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(uri).__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

__SCHEMES = ('ftp', 'ftps') class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

backend instance-attribute

rename(target)

Source code in src/uripath/schemes/_ftp.py

def rename(self, target: "FTPURIPath | URI | str"):
    # A plain str target is a sibling rename (relative to self's
    # parent), matching sftp.py's rename() semantics.
    target = self._rename_target(target)
    destination = self.with_path(target.path)
    core = self._core_path
    moved = core.with_segments(target.path or "/")
    with self._delegated((str(moved), str(destination))):
        core.rename(moved)
    # pathlib returns the new path.
    return destination

uripath.schemes.FTPURIBackend(timeout=DEFAULT_TIMEOUT, ssl_context=None, verify=True)

Bases: BaseFTPURIBackend

Connects via the stdlib FTP client (ftp:) or its TLS variant (ftps:, with PROT P for an encrypted data channel too).

timeout (seconds, default DEFAULT_TIMEOUT = 30) bounds connect, every reply and every transfer read; None blocks forever.

ftps: verifies the server certificate and host name by default (ssl.create_default_context()), before USER/PASS are sent. To trust a private CA or a self-signed certificate, pass your own ssl_context (e.g. ssl.create_default_context(cafile=...)). To turn verification off entirely -- accepting any certificate, so anyone on the network path can read the password -- pass verify=False. ssl_context wins over verify when both are given. Data connections reuse the control connection's TLS session.

Source code in src/uripath/schemes/_ftp.py

def __init__(
    self,
    timeout: "float | None" = DEFAULT_TIMEOUT,
    ssl_context: "_ssl.SSLContext | None" = None,
    verify: bool = True,
) -> None:
    self.timeout = timeout
    self.ssl_context = ssl_context
    self.verify = verify
    if not verify and ssl_context is None:
        _log.warning(
            "FTPURIBackend(verify=False): TLS certificates are not verified"
        )

__slots__ = ('timeout', 'ssl_context', 'verify') class-attribute instance-attribute

ssl_context = ssl_context instance-attribute

timeout = timeout instance-attribute

verify = verify instance-attribute

client(source, tls)

Source code in src/uripath/schemes/_ftp.py

def client(self, source: Source, tls: bool):
    user, password = source.parsed_userinfo()
    return FTPBackend(
        str(source.host),
        source.port or _netimps.get_default_port("ftp"),
        user=user,
        password=password,
        tls=tls,
        timeout=self.timeout,
        ssl_context=self.ssl_context,
        verify=self.verify,
    ).client()

uripath.schemes.BaseFTPURIBackend

Bases: object

Protocol for obtaining a connected+logged-in FTP client (or its TLS variant) for a Source. Subclass this to plug in custom connection handling (e.g. tests mock it directly, no real server); FTPURIBackend is the real implementation. A path to each endpoint (Source and tls) runs on a core pathapi.ftp.BaseFTPBackend that asks client() for its connections.

A context manager: leaving the block calls close(). One backend may be shared between threads; each thread keeps its own connection per server.

__slots__ = ('__weakref__',) class-attribute instance-attribute

__enter__()

Source code in src/uripath/schemes/_ftp.py

def __enter__(self):
    return self

__exit__(*exc_info)

Source code in src/uripath/schemes/_ftp.py

def __exit__(self, *exc_info) -> None:
    self.close()

client(source, tls)

The connected, logged-in client for source; tls asks for ftps:.

Source code in src/uripath/schemes/_ftp.py

@_notimplemented
def client(self, source: Source, tls: bool) -> "_ty.Any":
    """The connected, logged-in client for `source`; `tls` asks for `ftps:`."""

close()

Closes every control connection the paths of this backend hold, whichever thread opened it; one a command is running on closes when that command ends. Harmless when called again; a later use connects afresh.

Source code in src/uripath/schemes/_ftp.py

def close(self) -> None:
    """Closes every control connection the paths of this backend hold,
    whichever thread opened it; one a command is running on closes when
    that command ends. Harmless when called again; a later use connects
    afresh."""
    for endpoint in _endpoints_of(self):
        endpoint.close()