cdm_data_loaders.utils.file_transfer.ftp_client ¶

General-purpose FTP client utilities.

Provides resilient FTP connections with TCP keepalive, NOOP pings, retry on transient errors, and thread-local connection management for parallel downloads. Protocol-agnostic — callers supply the FTP hostname.

ThreadLocalFTP ¶

Manage thread-local FTP connections for parallel downloads.

Each thread gets its own FTP connection, created on first access. Call :meth:close_all when done to cleanly shut down all connections.

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

class ThreadLocalFTP:
    """Manage thread-local FTP connections for parallel downloads.

    Each thread gets its own FTP connection, created on first access.
    Call :meth:`close_all` when done to cleanly shut down all connections.
    """

    def __init__(self, host: str, timeout: int = DEFAULT_TIMEOUT) -> None:
        """Initialise with FTP host and timeout.

        :param host: FTP hostname (required — no default)
        :type host: str
        :param timeout: connection timeout in seconds
        :type timeout: int
        """
        self._host = host
        self._timeout = timeout
        self._local = threading.local()
        self._lock = threading.Lock()
        self._connections: list[FTP] = []

    def get(self) -> FTP:
        """Return the FTP connection for the current thread, reconnecting if stale.

        Sends a NOOP to verify the connection is still alive before returning it.
        If the server has closed the connection (e.g. after an idle timeout or
        session limit), the dead socket is discarded and a fresh connection is
        established transparently.
        """
        ftp = getattr(self._local, "ftp", None)
        if ftp is not None:
            try:
                ftp.voidcmd("NOOP")
            except Exception:  # noqa: BLE001
                # Connection is stale — discard it and reconnect below
                with contextlib.suppress(Exception):
                    ftp.quit()
                with self._lock, contextlib.suppress(ValueError):
                    self._connections.remove(ftp)
                ftp = None
                self._local.ftp = None
        if ftp is None:
            ftp = connect_ftp(self._host, self._timeout)
            self._local.ftp = ftp
            with self._lock:
                self._connections.append(ftp)
        return ftp

    def close_all(self) -> None:
        """Close all thread-local FTP connections."""
        with self._lock:
            for ftp in self._connections:
                with contextlib.suppress(Exception):
                    ftp.quit()
            self._connections.clear()

init ¶

__init__(host: str, timeout: int = DEFAULT_TIMEOUT) -> None

Initialise with FTP host and timeout.

Parameters:

Name	Type	Description	Default
`host`	`str`	FTP hostname (required — no default)	required
`timeout`	`int`	connection timeout in seconds	`DEFAULT_TIMEOUT`

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def __init__(self, host: str, timeout: int = DEFAULT_TIMEOUT) -> None:
    """Initialise with FTP host and timeout.

    :param host: FTP hostname (required — no default)
    :type host: str
    :param timeout: connection timeout in seconds
    :type timeout: int
    """
    self._host = host
    self._timeout = timeout
    self._local = threading.local()
    self._lock = threading.Lock()
    self._connections: list[FTP] = []

get ¶

get() -> FTP

Return the FTP connection for the current thread, reconnecting if stale.

Sends a NOOP to verify the connection is still alive before returning it. If the server has closed the connection (e.g. after an idle timeout or session limit), the dead socket is discarded and a fresh connection is established transparently.

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def get(self) -> FTP:
    """Return the FTP connection for the current thread, reconnecting if stale.

    Sends a NOOP to verify the connection is still alive before returning it.
    If the server has closed the connection (e.g. after an idle timeout or
    session limit), the dead socket is discarded and a fresh connection is
    established transparently.
    """
    ftp = getattr(self._local, "ftp", None)
    if ftp is not None:
        try:
            ftp.voidcmd("NOOP")
        except Exception:  # noqa: BLE001
            # Connection is stale — discard it and reconnect below
            with contextlib.suppress(Exception):
                ftp.quit()
            with self._lock, contextlib.suppress(ValueError):
                self._connections.remove(ftp)
            ftp = None
            self._local.ftp = None
    if ftp is None:
        ftp = connect_ftp(self._host, self._timeout)
        self._local.ftp = ftp
        with self._lock:
            self._connections.append(ftp)
    return ftp

close_all ¶

close_all() -> None

Close all thread-local FTP connections.

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def close_all(self) -> None:
    """Close all thread-local FTP connections."""
    with self._lock:
        for ftp in self._connections:
            with contextlib.suppress(Exception):
                ftp.quit()
        self._connections.clear()

connect_ftp ¶

connect_ftp(
    host: str, timeout: int = DEFAULT_TIMEOUT
) -> FTP

Connect and log in to an FTP server with TCP keepalive enabled.

Parameters:

Name	Type	Description	Default
`host`	`str`	FTP hostname	required
`timeout`	`int`	connection timeout in seconds	`DEFAULT_TIMEOUT`

Returns:

Type	Description
`FTP`	logged-in FTP connection

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def connect_ftp(host: str, timeout: int = DEFAULT_TIMEOUT) -> FTP:
    """Connect and log in to an FTP server with TCP keepalive enabled.

    :param host: FTP hostname
    :type host: str
    :param timeout: connection timeout in seconds
    :type timeout: int
    :return: logged-in FTP connection
    """
    ftp = FTP(host, timeout=timeout)  # noqa: S321
    ftp.login()
    _set_keepalive(ftp)
    return ftp

ftp_session ¶

ftp_session(
    host: str,
    ftp: FTP | None = None,
    timeout: int = DEFAULT_TIMEOUT,
) -> Iterator[FTP]

Context manager yielding an active FTP connection and closing it on exit if created here.

Parameters:

Name	Type	Description	Default
`host`	`str`	FTP hostname	required
`ftp`	`FTP \| None`	optional existing FTP connection (caller manages lifecycle when provided)	`None`
`timeout`	`int`	connection timeout in seconds	`DEFAULT_TIMEOUT`

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

@contextlib.contextmanager
def ftp_session(host: str, ftp: FTP | None = None, timeout: int = DEFAULT_TIMEOUT) -> Iterator[FTP]:
    """Context manager yielding an active FTP connection and closing it on exit if created here.

    :param host: FTP hostname
    :type host: str
    :param ftp: optional existing FTP connection (caller manages lifecycle when provided)
    :type ftp: FTP | None
    :param timeout: connection timeout in seconds
    :type timeout: int
    """
    owns_ftp = ftp is None
    active_ftp = connect_ftp(host, timeout=timeout) if ftp is None else ftp
    try:
        yield active_ftp
    finally:
        if owns_ftp:
            with contextlib.suppress(Exception):
                active_ftp.quit()

ftp_noop_keepalive ¶

ftp_noop_keepalive(
    ftp: FTP, last_activity: float, interval: int = 25
) -> float

Send NOOP if the connection has been idle longer than interval seconds.

Parameters:

Name	Type	Description	Default
`ftp`	`FTP`	active FTP connection	required
`last_activity`	`float`	monotonic timestamp of last FTP activity	required
`interval`	`int`	seconds of idle time before sending NOOP	`25`

Returns:

Type	Description
`float`	updated last-activity timestamp

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def ftp_noop_keepalive(ftp: FTP, last_activity: float, interval: int = 25) -> float:
    """Send NOOP if the connection has been idle longer than *interval* seconds.

    :param ftp: active FTP connection
    :type ftp: FTP
    :param last_activity: monotonic timestamp of last FTP activity
    :type last_activity: float
    :param interval: seconds of idle time before sending NOOP
    :type interval: int
    :return: updated last-activity timestamp
    """
    if time.monotonic() - last_activity > interval:
        with contextlib.suppress(Exception):
            ftp.sendcmd("NOOP")
        return time.monotonic()
    return last_activity

ftp_list_dir ¶

ftp_list_dir(
    ftp: FTP, path: PurePosixPath, retries: int = 3
) -> list[PurePosixPath]

List files in an FTP directory with retry on transient errors.

Parameters:

Name	Type	Description	Default
`ftp`	`FTP`	active FTP connection	required
`path`	`PurePosixPath`	remote directory path	required
`retries`	`int`	number of retry attempts	`3`

Returns:

Type	Description
`list[PurePosixPath]`	list of filenames

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def ftp_list_dir(ftp: FTP, path: PurePosixPath, retries: int = 3) -> list[PurePosixPath]:
    """List files in an FTP directory with retry on transient errors.

    :param ftp: active FTP connection
    :type ftp: FTP
    :param path: remote directory path
    :type path: PurePosixPath
    :param retries: number of retry attempts
    :type retries: int
    :return: list of filenames
    """
    ftp.cwd(str(path))

    @_ftp_retry(retries)
    def _list() -> list[PurePosixPath]:
        file_strings: list[str] = []
        ftp.retrlines("NLST", file_strings.append)
        return [PurePosixPath(f) for f in file_strings]

    return _list()

ftp_download_file ¶

ftp_download_file(
    ftp: FTP,
    remote_path: PurePosixPath,
    local_path: Path,
    retries: int = 3,
) -> None

Download a single file from FTP with retry on transient errors.

Parameters:

Name	Type	Description	Default
`ftp`	`FTP`	active FTP connection	required
`remote_path`	`PurePosixPath`	full remote file path	required
`local_path`	`Path`	local destination path	required
`retries`	`int`	number of retry attempts	`3`

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def ftp_download_file(ftp: FTP, remote_path: PurePosixPath, local_path: Path, retries: int = 3) -> None:
    """Download a single file from FTP with retry on transient errors.

    :param ftp: active FTP connection
    :type ftp: FTP
    :param remote_path: full remote file path
    :type remote_path: PurePosixPath
    :param local_path: local destination path
    :type local_path: Path
    :param retries: number of retry attempts
    :type retries: int
    """

    @_ftp_retry(retries)
    def _download() -> None:
        with local_path.open("wb") as f:
            ftp.retrbinary(f"RETR {remote_path}", f.write)

    _download()

ftp_retrieve_text ¶

ftp_retrieve_text(
    ftp: FTP, remote_path: PurePosixPath
) -> str

Retrieve a text file from FTP, returning its content as a string.

Parameters:

Name	Type	Description	Default
`ftp`	`FTP`	active FTP connection	required
`remote_path`	`PurePosixPath`	full remote file path	required

Returns:

Type	Description
`str`	file content

Source code in src/cdm_data_loaders/utils/file_transfer/ftp_client.py

def ftp_retrieve_text(ftp: FTP, remote_path: PurePosixPath) -> str:
    """Retrieve a text file from FTP, returning its content as a string.

    :param ftp: active FTP connection
    :type ftp: FTP
    :param remote_path: full remote file path
    :type remote_path: PurePosixPath
    :return: file content
    """
    lines: list[str] = []
    ftp.retrlines(f"RETR {remote_path}", lines.append)
    return "\n".join(lines)

cdm_data_loaders.utils.file_transfer.ftp_client ¶

ThreadLocalFTP ¶

__init__ ¶

get ¶

close_all ¶

connect_ftp ¶

ftp_session ¶

ftp_noop_keepalive ¶

ftp_list_dir ¶

ftp_download_file ¶

ftp_retrieve_text ¶

init ¶