Navigator / src / navigator / download / ftpserver / ftpfs_server.py
ftpfs_server.py
Raw
"""Implementation of the FTPFS server to be used by the download module."""

from pathlib import Path

from fs.ftpfs import FTPFS

__all__ = ["FTPFSServer"]


class FTPFSServer:
    """Implementation of the FTPFS server to be used by the download module.

    This class establishes a connection to an FTP server and provides methods
    to interact with the server, check its availability, and download files.

    Attributes:
        host (str): The FTP server hostname or IP address.
        user (str): The username for authentication.
        acct (str): The account name for authentication.
        tls (bool, optional): Indicates whether TLS encryption should be used. Defaults to True.

    Methods:
        __init__(self, host: str, user: str, acct: str, tls: bool = True) -> None:
            Initializes the FTPFSServer with the provided host, user, acct, and tls settings.

        connect(self) -> None:
            Establishes a connection to the FTP server using the specified credentials.

        is_alive(self) -> bool:
            Checks if the FTP server connection is alive by attempting to list the root directory.

        is_available(self, ftp_file_path: str) -> bool:
            Checks if a file or directory specified by ftp_file_path exists on the server.

        download(self, ftp_file_path: str, save_path: Path) -> None:
            Downloads a file from the FTP server identified by ftp_file_path and saves it to the local filesystem
            at the specified save_path.
    """

    def __init__(self, host: str, user: str, acct: str, tls: bool = True) -> None:
        """Initialize the FTPFSServer.

        Args:
            host (str): The FTP server hostname or IP address.
            user (str): The username for authentication.
            acct (str): The account name for authentication.
            tls (bool, optional): Indicates whether TLS encryption should be used. Defaults to True.
        """
        self.host = host
        self.user = user
        self.acct = acct
        self.tls = tls

    def connect(self) -> None:
        """Establish a connection to the FTP server using the specified credentials."""
        self.fs = FTPFS(
            host=self.host,
            user=self.user,
            acct=self.acct,
            tls=self.tls,
        )
        return

    def is_alive(self) -> bool:
        """Check if the FTP server connection is alive by attempting to list the root directory.

        Returns:
            bool: True if the connection is alive, False otherwise.
        """
        try:
            # Attempt to list the root directory
            self.fs.listdir("/")
            return True
        except Exception:
            return False

    def is_available(self, ftp_file_path: str) -> bool:
        """Check if a file or directory specified by ftp_file_path exists on the server.

        Args:
            ftp_file_path (str): The path of the file or directory on the server.

        Returns:
            bool: True if the file or directory exists, False otherwise.
        """
        if not self.is_alive():
            self.connect()

        return self.fs.exists(ftp_file_path)

    def download(self, ftp_file_path: str, save_path: str) -> None:
        """Download a file from the FTP server and save it to the local filesystem.

        Args:
            ftp_file_path (str): The path of the file on the FTP server.
            save_path (str): The path where the file should be saved on the local filesystem.
        """
        # If not connected, connect to the server
        if not self.is_alive():
            self.connect()

        # Get the file extension from the file path
        fname = Path(ftp_file_path).name

        # Open the file in binary mode
        with open(Path(save_path) / fname, "wb") as f:
            # Copy the file from the FTP server to the local file
            self.fs.download(ftp_file_path, f)

    def listdir(self, ftp_dir_path: str) -> list:
        """List the contents of a directory on the FTP server.

        Args:
            ftp_dir_path (str): The path of the directory on the FTP server.

        Returns:
            list: A list of the contents of the directory.
        """
        # If not connected, connect to the server
        if not self.is_alive():
            self.connect()

        # List the contents of the directory
        return self.fs.listdir(ftp_dir_path)

    def close(self) -> None:
        """Close the connection to the FTP server."""
        if self.is_alive():
            self.fs.close()
        return