FTP Path API

pathapi.ftp

FTP and FTPS paths on the standard library alone (ftplib).

FTPPath is a Path for a directory tree on an FTP server and needs no URI; FTPBackend holds the connections to one server and account.

DEFAULT_PORT = 21 module-attribute

The control port FTPBackend connects to when it is not given one. tls asks for explicit FTPS on that same port; port 990 is never assumed.

DEFAULT_TIMEOUT = 30.0 module-attribute

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

IDLE_PROBE_SECONDS = 1.0 module-attribute

A cached control connection that carried a command within this many seconds is used as it is. One that has been idle longer, or has not carried a command yet, is probed with NOOP first and replaced if the server dropped it. Servers close idle sessions after tens of seconds at the least, so a run of operations pays no probe and a pause long enough to lose the session pays one.

BaseFTPBackend

Bases: object

Contract for obtaining a connected, logged-in ftplib.FTP (or FTP_TLS). Subclass this to plug in custom connection handling (tests mock it directly, no real server); FTPBackend is the real implementation. endpoint, when a subclass sets it, is a hashable value naming the server and account: two backends with equal endpoints serve the same files.

endpoint = None class-attribute instance-attribute

client()

A new connected, logged-in client.

Source code in src/pathapi/ftp/_backend.py

@_notimplemented
def client(self) -> "_ftplib.FTP":
    """A new connected, logged-in client."""

close()

Closes every control connection this backend holds, 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/pathapi/ftp/_backend.py

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

FTPBackend(host, port=DEFAULT_PORT, *, user=None, password=None, tls=False, timeout=DEFAULT_TIMEOUT, ssl_context=None, verify=True)

Bases: BaseFTPBackend

The connections to one FTP server, as one account: stdlib ftplib.FTP, or ftplib.FTP_TLS with tls=True (explicit FTPS on the control port, PROT P for an encrypted data channel too).

Nothing connects when the backend is built. Each thread opens a control connection on its first use and keeps it, so paths that share a backend share a connection per thread (close() ends them, and so does the end of the thread). timeout (seconds, default DEFAULT_TIMEOUT = 30) bounds connect, every reply and every transfer read; None blocks forever. Without user the login is anonymous.

tls=True 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.

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

Source code in src/pathapi/ftp/_backend.py

def __init__(
    self,
    host: str,
    port: int = DEFAULT_PORT,
    *,
    user: "str | None" = None,
    password: "str | None" = None,
    tls: bool = False,
    timeout: "float | None" = DEFAULT_TIMEOUT,
    ssl_context: "_ssl.SSLContext | None" = None,
    verify: bool = True,
) -> None:
    if not host:
        raise ValueError("an FTP backend needs a host")
    self.host = str(host)
    self.port = DEFAULT_PORT if port is None else int(port)
    self.user = user
    self.password = password
    self.tls = bool(tls)
    self.timeout = timeout
    self.ssl_context = ssl_context
    self.verify = verify

endpoint property

host = str(host) instance-attribute

password = password instance-attribute

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

ssl_context = ssl_context instance-attribute

timeout = timeout instance-attribute

tls = bool(tls) instance-attribute

user = user instance-attribute

verify = verify instance-attribute

client()

Source code in src/pathapi/ftp/_backend.py

def client(self) -> "_ftplib.FTP":
    if self.tls:
        client = _SessionReuseFTP_TLS(
            context=self._tls_context(), timeout=self.timeout
        )
    else:
        client = _ftplib.FTP(timeout=self.timeout)
    try:
        client.connect(self.host, self.port)
        client.login(self.user or "anonymous", self.password or "")
        if self.tls:
            client.prot_p()
        client.set_pasv(True)
    except BaseException:
        # A rejected certificate or a timed-out greeting must not leave
        # the half-open control socket behind.
        close = getattr(client, "close", None)
        if close is not None:
            close()
        raise
    return client

FTPEndpoint

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.

host instance-attribute

port instance-attribute

tls instance-attribute

user instance-attribute

FTPPath(*segments, backend=None, host=None, port=DEFAULT_PORT, user=None, password=None, tls=False, timeout=DEFAULT_TIMEOUT, ssl_context=None, verify=True)

Bases: TreePath

A path on an FTP server, read and written through stdlib ftplib, with a thread-keyed connection cache (FTPBackend; 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).

FTPPath("/pub/x", host="h", user="u", tls=True) builds the path's own FTPBackend; backend= shares one connection cache between paths built separately, or plugs a BaseFTPBackend of your own. A path joined from another keeps its backend. Names are read from the server root: pub/x and /pub/x are one file.

Source code in src/pathapi/ftp/_path.py

def __init__(
    self,
    *segments: "str | FTPPath",
    backend: "BaseFTPBackend | None" = None,
    host: "str | None" = None,
    port: int = DEFAULT_PORT,
    user: "str | None" = None,
    password: "str | None" = None,
    tls: bool = False,
    timeout: "float | None" = DEFAULT_TIMEOUT,
    ssl_context: "object | None" = None,
    verify: bool = True,
):
    text, donor = self._join(segments)
    endpoint = (
        host is not None
        or user is not None
        or password is not None
        or ssl_context is not None
        or tls
        or port != DEFAULT_PORT
        or timeout != DEFAULT_TIMEOUT
        or verify is not True
    )
    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 = FTPBackend(
                host,
                port,
                user=user,
                password=password,
                tls=tls,
                timeout=timeout,
                ssl_context=ssl_context,
                verify=verify,
            )
        elif donor is not None:
            backend = donor._backend
        else:
            raise TypeError("an FTPPath needs host= or backend=")
    self._segments = self._parse(text)
    self._backend = backend

backend property

The FTPBackend (or custom BaseFTPBackend) holding the connections of this path.

as_uri()

Source code in src/pathapi/ftp/_path.py

def as_uri(self) -> str:
    backend = self._backend
    host = getattr(backend, "host", None)
    if host is None:
        raise NotImplementedError("as_uri() needs an FTPBackend")
    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 != DEFAULT_PORT:
        netloc = f"{netloc}:{backend.port}"
    scheme = "ftps" if backend.tls else "ftp"
    return f"{scheme}://{netloc}{_urlparse.quote(self._wirepath, safe='/')}"

chmod(mode, *, follow_symlinks=True)

Source code in src/pathapi/ftp/_path.py

def chmod(self, mode: "int | str", *, follow_symlinks: bool = True):
    # SITE CHMOD is a non-standard extension that many servers lack. A rejection of an
    # existing path becomes NotImplementedError, which copy() skips; a missing path is
    # FileNotFoundError.
    if not follow_symlinks:
        raise NotImplementedError("chmod(follow_symlinks=False)")
    mode = _parse_mode(mode)
    try:
        self._call("voidcmd", f"SITE CHMOD {mode:o} {self._wirepath}")
    except _ftplib.error_perm as error:
        if reply_code(error) in UNSUPPORTED_REPLIES:
            raise NotImplementedError(
                f"SITE CHMOD not supported by this server ({error})"
            ) from error
        translated = self._translate(error)
        if isinstance(translated, FileNotFoundError):
            raise translated from error
        raise NotImplementedError(
            f"SITE CHMOD refused by the server ({error})"
        ) from error

iterdir()

Source code in src/pathapi/ftp/_path.py

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

rename(target)

Source code in src/pathapi/ftp/_path.py

def rename(self, target: "FTPPath | str"):
    # A plain str target is a sibling rename (relative to self's
    # parent), matching sftp.py's rename() semantics.
    if isinstance(target, str):
        target = self.parent / target
    elif not isinstance(target, FTPPath) or not self._same_filesystem(target):
        raise NotImplementedError("rename() onto another server")
    destination = self._rebuild(target.as_posix())
    try:
        self._call("rename", self._wirepath, destination._wirepath)
    except _ftplib.error_perm as error:
        raise self._rename_error(error, destination) from error
    # pathlib returns the new path.
    return destination

rmdir()

Source code in src/pathapi/ftp/_path.py

def rmdir(self):
    try:
        self._call("rmd", self._wirepath)
    except _ftplib.error_perm as error:
        raise self._translate(error, self._rmdir_mismatch(error)) from error

stat(*, follow_symlinks=True)

Source code in src/pathapi/ftp/_path.py

def stat(self, *, follow_symlinks=True):
    return self._fresh_stat()

unlink(missing_ok=False)

Source code in src/pathapi/ftp/_path.py

def unlink(self, missing_ok=False):
    try:
        self._call("delete", self._wirepath)
    except _ftplib.error_perm as error:
        translated = self._translate(error, self._is_a_directory)
        if missing_ok and isinstance(translated, FileNotFoundError):
            return
        raise translated from error