Paramiko fetch files from sFTP server
Here's the refactored version of the function with that check added:
import paramiko
from pathlib import Path
import time
from typing import Tuple, Optional
def fetch_sftp_file(
username: str,
password: str,
sftp_host: str,
server_file_path: str,
server_filename: str,
local_file_path: Optional[str] = None,
local_filename: Optional[str] = None,
ssh_connection_timeout: float = 10.0,
retry_count: int = 3,
retry_delay: int = 2,
sftp_transfer_timeout: float = 60.0,
) -> Tuple[str, str]:
"""
Fetches a file from an SFTP server and saves it locally, with retry logic for both SSH connection and SFTP file transfer errors.
Parameters:
username: SFTP username
password: SFTP password
sftp_host: SFTP server hostname or IP
server_file_path: Path on SFTP server where the file is located
server_filename: Name of the file on the SFTP server
local_file_path: Local directory to save the file (default: current working directory)
local_filename: Name of the file to save locally (default: same as server filename)
ssh_connection_timeout: SSH connection timeout (default: 10 seconds)
retry_count: Number of retry attempts for SSH/SFTP (default: 3)
retry_delay: Delay between retry attempts (default: 2 seconds)
sftp_transfer_timeout: Timeout for file transfer (default: 60 seconds)
Returns:
Tuple[str, str]: On success, returns ("success", absolute local file path)
Tuple[str, str]: On failure, returns ("failure", error message + exception info)
"""
# Validate input parameters
# Edge case 1: Validate input types
if not isinstance(server_file_path, str):
return ("failure", f"server_file_path is not a string. It is of type: {type(server_file_path)}")
if not isinstance(server_filename, str):
return ("failure", f"server_filename is not a string. It is of type: {type(server_filename)}")
# Edge case 2: Handle empty input strings
if not server_file_path.strip() or not server_file_path.strip() or not local_file_path.strip() or not local_filename.strip():
return ("failure", "directory_path and filename cannot be empty or just whitespace. It is of type: {type(server_filename)}")
remote_path = str(Path(server_file_path).resolve(strict=False).as_posix() / Path(server_filename))
if local_filename is None:
local_file = Path(server_filename)
else:
local_file = Path(local_filename)
if local_file_path is None:
local_dir = Path.cwd()
else:
# Check if the local directory exists and is a directory
if not local_file_path.is_dir():
return ("failure", f"'{local_file_path}' is not a directory.")
if not local_file_path.exists():
return ("failure", f"Local directory '{local_file_path}' does not exist.")
else:
local_dir = Path(local_file_path).resolve()
local_path = str(local_dir / local_file)
for ssh_attempt in range(retry_count):
try:
with paramiko.SSHClient() as ssh:
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=sftp_host, username=username, password=password, timeout=ssh_connection_timeout)
with ssh.open_sftp() as sftp:
sftp.settimeout(sftp_transfer_timeout)
# Attempt SFTP file transfer with retry
for transfer_attempt in range(retry_count):
try:
sftp.get(remote_path, local_path)
return ("success", local_path)
except paramiko.SFTPError as e:
if transfer_attempt < retry_count - 1:
time.sleep(retry_delay)
continue
return ("failure", f"SFTP transfer failed: {e} (code: {e.code})")
except PermissionError as e:
return ("failure", f"Local file operation failed: {e}")
except Exception as e:
if transfer_attempt < retry_count - 1:
time.sleep(retry_delay)
continue
return ("failure", f"Unexpected error during SFTP transfer: {e}")
# If transfer retries exhausted
return ("failure", "SFTP transfer retries exhausted after multiple attempts")
except paramiko.AuthenticationException as e:
if ssh_attempt < retry_count - 1:
time.sleep(retry_delay)
continue
return ("failure", f"Authentication failed: {e}")
except paramiko.SSHException as e:
if ssh_attempt < retry_count - 1:
time.sleep(retry_delay)
continue
return ("failure", f"SSH connection failed: {e}")
except Exception as e:
if ssh_attempt < retry_count - 1:
time.sleep(retry_delay)
continue
return ("failure", f"Unexpected error during SSH connection: {e}")
return ("failure", "All retry attempts failed to connect to SFTP server")