FTP

FtpConnProperties(host, user=None, password=None, timeout=60, passive=True) dataclass

Class that holds the properties used to establish a connection to an FTP server.

Parameters:

• host

  (str) –

  The hostname of the database server.

• user

  (str | None, default: None ) –

  User name. By default None.

• password

  (str | None, default: None ) –

  Password for the user. By default None.

• timeout

  (int, default: 60 ) –

  Timeout in seconds. By default 60.

• passive

  (bool, default: True ) –

  If set to true will use PASV command to use a passive connection (usually better when behind firewalls). By default True.

__post_init__()

Method that checks if inputs are valid after initialization.

Source code in echo_connhandler/ftp_core.py

def __post_init__(self) -> None:
    """Method that checks if inputs are valid after initialization."""
    if not isinstance(self.host, str):
        raise TypeError(f"host must be a string not {type(self.host)}")
    if not isinstance(self.user, str | type(None)):
        raise TypeError(f"user must be a string or None not {type(self.user)}")
    if not isinstance(self.password, str | type(None)):
        raise TypeError(f"password must be a string or None not {type(self.password)}")
    if not isinstance(self.timeout, int):
        raise TypeError(f"timeout must be an integer not {type(self.timeout)}")
    if not isinstance(self.passive, bool):
        raise TypeError(f"passive must be a boolean not {type(self.passive)}")

FtpHandler(connection_properties, max_retries=8, max_conn_retries=3, retry_wait_time=30, exponential_min_retry_wait_time=1, exponential_multiplier=1, skip_connect=False, **kwargs)

Base abstract class for FTP handlers.

This class should be used to handle connecting to FTP servers.

This already connects to the FTP server.

This handler implements a robust retry strategy using exponential backoff to gracefully manage connection or query failures. By default, operations are attempted up to 8 times; between failures, the wait time doubles exponentially—starting at 1 second (1s, 2s, 4s, 8s, 16s)—until it reaches a configured ceiling of 30 seconds. This approach allows for rapid recovery from momentary glitches while preventing server overload during persistent outages.

Using the default parameters the connection will fail after around 90 seconds if the server is not reachable.

Parameters:

• connection_properties

  (FtpConnProperties) –

  Object containing connection parameters.

• max_retries

  (int, default: 8 ) –

  Number of retries that will be attempted when doing queries. Will be used in stop parameter of tenacity.stop_after_attempt, by default 8

• max_conn_retries

  (int, default: 3 ) –

  Number of retries that will be attempted when reconnecting. Will be used in stop parameter of tenacity.stop_after_attempt, by default 3

• retry_wait_time

  (float, default: 30 ) –

  Max time to wait between retries when reconnecting or doing queries. Will be used in max parameter of tenacity.wait_exponential, by default 30

• exponential_min_retry_wait_time

  (float, default: 1 ) –

  Min time to wait between retries when reconnecting or doing queries. Will be used in min parameter of tenacity.wait_exponential, by default 1

• exponential_multiplier

  (float, default: 1 ) –

  Multiplier to use when calculating wait time between retries when reconnecting or doing queries. Will be used in multiplier parameter of tenacity.wait_exponential, by default 1

• skip_connect

  (bool, default: False ) –

  If True, the connection will not be established when the object is created.

  If this is set toTrue, the user will need to manually call the reconnect() method when they want to connect to the server.

  By default False

• **kwargs

  (dict, default: {} ) –

  Just kept here for compatibility.

Source code in echo_connhandler/ftp_core.py

@validate_call
def __init__(
    self,
    connection_properties: FtpConnProperties,
    max_retries: int = 8,
    max_conn_retries: int = 3,
    retry_wait_time: float = 30,
    exponential_min_retry_wait_time: float = 1,
    exponential_multiplier: float = 1,
    skip_connect: bool = False,
    **kwargs,  # pylint: disable=unused-argument # noqa
) -> None:
    """Method that initializes the FTP handler.

    This already connects to the FTP server.

    This handler implements a robust retry strategy using exponential backoff to gracefully manage connection or query failures. By default, operations are attempted up to 8 times; between failures, the wait time doubles exponentially—starting at 1 second (1s, 2s, 4s, 8s, 16s)—until it reaches a configured ceiling of 30 seconds. This approach allows for rapid recovery from momentary glitches while preventing server overload during persistent outages.

    Using the default parameters the connection will fail after around 90 seconds if the server is not reachable.

    Parameters
    ----------
    connection_properties : FtpConnProperties
        Object containing connection parameters.
    max_retries : int, optional
        Number of retries that will be attempted when doing queries. Will be used in `stop` parameter of tenacity.stop_after_attempt, by default 8
    max_conn_retries : int, optional
        Number of retries that will be attempted when reconnecting. Will be used in `stop` parameter of tenacity.stop_after_attempt, by default 3
    retry_wait_time : float, optional
        Max time to wait between retries when reconnecting or doing queries. Will be used in `max` parameter of tenacity.wait_exponential, by default 30
    exponential_min_retry_wait_time: float, optional
        Min time to wait between retries when reconnecting or doing queries. Will be used in `min` parameter of tenacity.wait_exponential, by default 1
    exponential_multiplier: float, optional
        Multiplier to use when calculating wait time between retries when reconnecting or doing queries. Will be used in `multiplier` parameter of tenacity.wait_exponential, by default 1
    skip_connect : bool, optional
        If True, the connection will not be established when the object is created.

        If this is set toTrue, the user will need to manually call the reconnect() method when they want to connect to the server.

        By default False
    **kwargs : dict, optional
        Just kept here for compatibility.

    """
    self._cwd = "/"
    super().__init__(
        connection_properties=connection_properties,
        max_retries=max_retries,
        max_conn_retries=max_conn_retries,
        retry_wait_time=retry_wait_time,
        exponential_min_retry_wait_time=exponential_min_retry_wait_time,
        exponential_multiplier=exponential_multiplier,
        skip_connect=skip_connect,
    )

change_directory(dirname)

Function to change the current working directory.

Parameters:

• dirname

  (str) –

  New desired working directory.

Returns:

• str –

  Output of command. Not relevant.

Source code in echo_connhandler/ftp_core.py

@validate_call
def change_directory(self, dirname: str) -> str:
    """Function to change the current working directory.

    Parameters
    ----------
    dirname : str
        New desired working directory.

    Returns
    -------
    str
        Output of command. Not relevant.

    """
    result = self._ftp_command(self.connection.cwd, dirname=dirname)
    # saving this directory as cwd
    self._cwd = self.current_directory()

    return result

current_directory()

Function to get the current working directory.

Returns:

• str –

  Current working directory.

Source code in echo_connhandler/ftp_core.py

def current_directory(self) -> str:
    """Function to get the current working directory.

    Returns
    -------
    str
        Current working directory.

    """
    return self._ftp_command(self.connection.pwd)

delete_file(filename)

Function to delete a file.

Parameters:

• filename

  (str) –

  File to be deleted.

Returns:

• str –

  Output of command. Not relevant.

Source code in echo_connhandler/ftp_core.py

@validate_call
def delete_file(self, filename: str) -> None:
    """Function to delete a file.

    Parameters
    ----------
    filename : str
        File to be deleted.

    Returns
    -------
    str
        Output of command. Not relevant.

    """
    return self._ftp_command(self.connection.delete, filename=filename)

get_file(filename, dest_directory)

Function to download a file.

Parameters:

• filename

  (str) –

  Name of the file to be downloaded.

• dest_directory

  (Path) –

  Directory where the file will be downloaded.

Source code in echo_connhandler/ftp_core.py

@validate_call
def get_file(self, filename: str, dest_directory: Path) -> None:
    """Function to download a file.

    Parameters
    ----------
    filename : str
        Name of the file to be downloaded.
    dest_directory : Path
        Directory where the file will be downloaded.

    """
    # creating dest folder if it does not exist
    Path(dest_directory).mkdir(parents=True, exist_ok=True)

    # command to copy file
    command: str = f"RETR {filename}"

    # executing command (chunks of 1 MB) with proper file handling
    dest_file = Path(f"{dest_directory}/{filename}")
    with dest_file.open("wb") as f:
        self._ftp_command(
            self.connection.retrbinary,
            cmd=command,
            callback=f.write,
            blocksize=1 * 1024 * 1024,
        )

list_contents()

Function to list all files in the current directory.

Returns:

• list[str] –

  List of all files/dirs in the current directory.

Source code in echo_connhandler/ftp_core.py

def list_contents(self) -> list[str]:
    """Function to list all files in the current directory.

    Returns
    -------
    list[str]
        List of all files/dirs in the current directory.

    """
    return self._ftp_command(self.connection.nlst)

make_directory(dirname)

Function to create a directory.

Parameters:

• dirname

  (str) –

  New desired directory.

Returns:

• str –

  Output of command. Not relevant.

Source code in echo_connhandler/ftp_core.py

@validate_call
def make_directory(self, dirname: str) -> None:
    """Function to create a directory.

    Parameters
    ----------
    dirname : str
        New desired directory.

    Returns
    -------
    str
        Output of command. Not relevant.

    """
    return self._ftp_command(self.connection.mkd, dirname=dirname)

remove_directory(dirname)

Function to remove a directory.

Parameters:

• dirname

  (str) –

  Directory to be removed.

Returns:

• str –

  Output of command. Not relevant.

Source code in echo_connhandler/ftp_core.py

@validate_call
def remove_directory(self, dirname: str) -> None:
    """Function to remove a directory.

    Parameters
    ----------
    dirname : str
        Directory to be removed.

    Returns
    -------
    str
        Output of command. Not relevant.

    """
    return self._ftp_command(self.connection.rmd, dirname=dirname)

rename_file(fromname, toname)

Function to rename a file.

Parameters:

• fromname

  (str) –

  Original name of the file.

• toname

  (str) –

  New name of the file.

Returns:

• str –

  Output of command. Not relevant.

Source code in echo_connhandler/ftp_core.py

@validate_call
def rename_file(self, fromname: str, toname: str) -> None:
    """Function to rename a file.

    Parameters
    ----------
    fromname : str
        Original name of the file.
    toname : str
        New name of the file.

    Returns
    -------
    str
        Output of command. Not relevant.

    """
    return self._ftp_command(self.connection.rename, fromname=fromname, toname=toname)