Add timeout parameter and adjust directory handling

Author: vinod-kanigicherla

tasks/ftp/retrieve_from_ftp.py

@@ -16,6 +16,7 @@ def __init__(
         username: str,
         password: str,
         local_dir: str = "./",
+        timeout: int = 30,
         *args,
         **kwargs,
     ):
@@ -28,6 +29,7 @@ def __init__(
           username (str): Username for FTP authentication.
           password (str): Password for FTP authentication.
           local_dir (str, optional): Local directory to save the retrieved file. Defaults to "./".
+          timeout (int, optional): Timeout value for FTP connection measued in seconds. Defaults to 30 seconds.
           *args: Variable length argument list.
           **kwargs: Arbitrary keyword arguments.
         """
@@ -37,6 +39,7 @@ def __init__(
         self.username = username
         self.password = password
         self.local_dir = local_dir
+        self.timeout = timeout
 
     def run(self) -> Result:
         """
@@ -55,16 +58,16 @@ def run(self) -> Result:
 
         """
         try:
-            ftp = FTP(self.ftp_url, timeout=30)  # Modify timeout as needed
+            if not os.path.isdir(self.local_dir):
+                os.makedirs(self.local_dir, exist_ok=True)
+
+            ftp = FTP(self.ftp_url, timeout=self.timeout)  # Modify timeout as needed
             ftp.login(user=self.username, passwd=self.password)
 
             remote_dir, remote_filename = os.path.split(self.ftp_remote_filepath)
             if remote_dir:
                 ftp.cwd(remote_dir)
 
-            if not os.path.isdir(self.local_dir):
-                return Failure(f"Local directory does not exist: {self.local_dir}")
-
             local_filepath = os.path.join(self.local_dir, remote_filename)
             with open(local_filepath, "wb") as f:
                 ftp.retrbinary(f"RETR {remote_filename}", f.write)

tasks/ftp/upload_to_ftp.py

@@ -16,6 +16,7 @@ def __init__(
         username: str,
         password: str,
         destination_dir: str = "/",
+        timeout: int = 30,
         *args,
         **kwargs,
     ):
@@ -28,13 +29,15 @@ def __init__(
           username (str): Username credential for FTP auth.
           password (str): Password credential for FTP auth.
           destination_dir (str): Destination directory on the FTP server where the file will be uploaded to. Defaults to "/".
+          timeout (int, optional): Timeout value for FTP connection measued in seconds. Defaults to 30 seconds.
         """
         super().__init__(*args, **kwargs)
         self.local_filepath = local_filepath
         self.ftp_url = ftp_url
         self.username = username
         self.password = password
         self.destination_dir = destination_dir
+        self.timeout = timeout
 
     def run(self) -> Result:
         """
@@ -57,7 +60,7 @@ def run(self) -> Result:
             if not os.path.isfile(self.local_filepath):
                 return Failure(f"Local file does not exist: {self.local_filepath}")
 
-            ftp = FTP(self.ftp_url, timeout=30)  # Modify timeout as needed
+            ftp = FTP(self.ftp_url, timeout=self.timeout)  # Modify timeout as needed
             ftp.login(user=self.username, passwd=self.password)
 
             if self.destination_dir: