ftp: and ftps:

pathlib_next.uri.schemes.ftp

DEFAULT_TIMEOUT = 30.0 module-attribute

Socket timeout, in seconds, FtpBackend applies to connect, replies and transfers when the caller does not pass one.

BaseFtpBackend

Bases: object

Protocol for obtaining a connected+logged-in ftplib.FTP (or FTP_TLS) for a Source. Subclass this to plug in custom connection handling (e.g. tests mock it directly, no real server); FtpBackend is the real implementation.

FtpBackend(timeout=DEFAULT_TIMEOUT, ssl_context=None, verify=True)

Bases: BaseFtpBackend

Connects via stdlib ftplib.FTP (ftp:) or ftplib.FTP_TLS (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/pathlib_next/uri/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

FtpPath(*uris, **options)

Bases: UriPath

ftp:/ftps: scheme: full read/write access via stdlib ftplib, with a thread-keyed LRU connection cache (_CACHED_CLIENTS, mirroring sftp.py). Directory listing and stat prefer MLSD (RFC 3659 -- gives type/size/modify facts in one round trip); servers that don't support it fall back to NLST for listing (names only) and SIZE for file stat (no portable "not found vs. is a directory" distinction in that path).

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