# -*- coding: utf-8 -*-
"""
SFTP Client Module
==================
This module provides a high-level SFTP client wrapper around Paramiko for secure file operations.
Key Features:
- Simplified SFTP connection management
- Comprehensive error handling with custom exceptions
- Context manager support for automatic resource cleanup
- Support for both password and key-based authentication
- File upload, download, listing, and deletion operations
Example:
>>> from core_ftp.clients.sftp import SftpClient, SftpConnectionConfig, SftpTransportConfig
>>>
>>> config = SftpConnectionConfig(host="server.com", user="admin", password="secret")
>>> with SftpClient(config) as client:
... files = list(client.list_files("/data"))
... client.download_file("remote.txt", "local.txt")
"""
from dataclasses import dataclass
from typing import (
Any,
Callable,
cast,
Dict,
IO,
Iterator,
List,
Optional,
Tuple,
)
from core_mixins.compatibility import Self
from paramiko import Transport, RSAKey
from paramiko.sftp_attr import SFTPAttributes
from paramiko.sftp_client import SFTPClient
from paramiko.ssh_exception import AuthenticationException
from paramiko.ssh_exception import BadHostKeyException
from paramiko.ssh_exception import NoValidConnectionsError
from paramiko.ssh_exception import SSHException
[docs]
@dataclass
class SftpConnectionConfig:
"""Configuration for establishing an SFTP connection."""
host: str
port: int = 22
user: Optional[str] = None
password: Optional[str] = None
private_key_path: str = ""
passphrase: Optional[str] = None
[docs]
@dataclass
class SftpTransportConfig:
"""Low-level Paramiko transport/connection options."""
transport_kwargs: Optional[Dict[str, Any]] = None
connection_kwargs: Optional[Dict[str, Any]] = None
disabled_algorithms: bool = False
algorithms_to_disable: Optional[List[str]] = None
[docs]
class SftpClient:
"""
High-level SFTP client that wraps Paramiko for secure file transfer operations.
.. code-block:: python
conn = SftpConnectionConfig(host="test.rebex.net", user="demo", password="password")
client = SftpClient(conn)
client.connect()
for x in client.list_files("/"):
print(x)
client.close()
with SftpClient(conn) as _client:
_client.download_file("readme.txt", "/tmp/readme.txt")
..
"""
[docs]
def __init__(
self,
connection: SftpConnectionConfig,
transport: Optional[SftpTransportConfig] = None,
) -> None:
"""
Initialize SFTP client with connection parameters.
:param connection: SFTP connection configuration.
:type connection: SftpConnectionConfig
:param transport: Low-level Paramiko transport options.
:type transport: Optional[SftpTransportConfig]
"""
self.connection = connection
self.transport = transport if transport is not None else SftpTransportConfig()
self._sftp_client: Optional[SFTPClient] = None
self._transport: Optional[Transport] = None
@property
def client(self) -> SFTPClient:
"""
Provides access to the underlying Paramiko SFTP client.
Auto-connects if not already connected.
:return: The underlying `SFTPClient` instance.
:rtype: SFTPClient
:raises SftpClientError: If connection fails.
"""
if self._sftp_client is None:
self.connect()
return cast(SFTPClient, self._sftp_client)
[docs]
def _ensure_transport(self) -> Transport:
"""
Ensures transport connection exists, creating it if necessary.
:return: The transport instance.
:rtype: `Transport`.
"""
if self._transport is None:
transport_kwargs = dict(self.transport.transport_kwargs or {})
# It's a bug in Paramiko. It does not handle correctly absence
# of server-sig-algs extension on the server side...
# https://stackoverflow.com/questions/70565357/paramiko-authentication-fails-with-agreed-upon-rsa-sha2-512-pubkey-algorithm
if self.transport.disabled_algorithms:
transport_kwargs["disabled_algorithms"] = {
"pubkeys": (
self.transport.algorithms_to_disable or ["rsa-sha2-512", "rsa-sha2-256"]
)
}
self._transport = Transport(
(self.connection.host, self.connection.port),
**transport_kwargs,
)
return self._transport
def __enter__(self) -> Self:
"""
Context manager entry point.
:return: The SFTP client instance.
:rtype: Self
"""
self.connect()
return self
[docs]
def connect(self) -> Self:
"""
Establishes SFTP connection to the remote server.
:return: The SFTP client instance for method chaining.
:rtype: Self
:raises SftpClientError:
If connection fails due to authentication,
host key, SSH, or other errors.
"""
data: Dict[str, Any] = {
"username": self.connection.user,
"password": self.connection.password,
}
try:
if self.connection.private_key_path:
data["pkey"] = RSAKey.from_private_key_file(
self.connection.private_key_path,
self.connection.passphrase,
)
_transport = self._ensure_transport()
_transport.connect(**data, **(self.transport.connection_kwargs or {}))
self._sftp_client = SFTPClient.from_transport(_transport)
return self
except AuthenticationException as error:
raise SftpClientError(f"Authentication error: {error}.") from error
except BadHostKeyException as error:
raise SftpClientError(f"HostKeys error: {error}.") from error
except SSHException as error:
raise SftpClientError(f"SSH error: {error}.") from error
except NoValidConnectionsError as error:
raise SftpClientError(f"Connection error: {error}") from error
except Exception as error:
raise SftpClientError(f"Error: {error}.") from error
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""
Context manager exit point.
:param exc_type: Exception type.
:param exc_val: Exception value.
:param exc_tb: Exception traceback.
"""
self.close()
[docs]
def close(self) -> None:
"""
Closes the SFTP connection and transport.
Safe to call multiple times or on unopened connections.
"""
if self._sftp_client is not None:
self._sftp_client.close()
if self._transport is not None:
self._transport.close()
[docs]
def get_cwd(self) -> Optional[str]:
"""
Returns the current working directory on the remote server.
:return: Current working directory path, or None if not set.
:rtype: Optional[str]
:raises SftpClientError: If unable to get current directory.
"""
try:
return self.client.getcwd()
except IOError as error:
raise SftpClientError(f"Error getting current directory: {error}") from error
[docs]
def chdir(self, remote_path: str) -> None:
"""
Changes the current working directory on the remote server.
:param remote_path: Path to change to.
:type remote_path: str
:raises SftpClientError: If unable to change directory.
"""
try:
self.client.chdir(remote_path)
except IOError as error:
raise SftpClientError(f"Error changing directory: {error}") from error
[docs]
def list_files(self, remote_path: str) -> Iterator[Tuple[str, SFTPAttributes]]:
"""
Read files under a remote directory.
:param remote_path: Remote directory path.
:return: Iterator of tuples in the form ("file_name", SFTPAttributes)
"""
try:
for attr in self.client.listdir_attr(remote_path):
yield attr.filename, attr
except IOError as error:
raise SftpClientError(f"Error accessing directory: {error}") from error
[docs]
def download_file(self, remote_file_path: str, local_file_path: str) -> str:
"""
Downloads a file from the remote server to local filesystem.
:param remote_file_path: Path to the remote file.
:type remote_file_path: str
:param local_file_path: Local path where file will be saved.
:type local_file_path: str
:return: The local file path where file was saved.
:rtype: str
:raises SftpClientError: If download fails.
"""
try:
self.client.get(remote_file_path, local_file_path)
return local_file_path
except IOError as error:
raise SftpClientError(f"Error downloading file: {error}") from error
[docs]
def upload_file(
self,
file_path: str,
remote_path: str,
callback: Optional[Callable[[int, int], Any]] = None,
confirm: bool = False,
) -> SFTPAttributes:
"""
Uploads a local file to the remote server.
:param file_path: Local path to the file to upload.
:type file_path: str
:param remote_path: Remote path where file will be stored.
:type remote_path: str
:param callback: Optional callback for progress monitoring.
:type callback: Optional[Callable[[int, int], Any]]
:param confirm: Whether to confirm the upload.
:type confirm: bool
:return: File attributes of the uploaded file.
:rtype: SFTPAttributes
:raises SftpClientError: If upload fails.
"""
try:
return self.client.put(
file_path,
remotepath=remote_path,
callback=callback,
confirm=confirm,
)
except IOError as error:
raise SftpClientError(f"Error uploading file: {error}") from error
[docs]
def upload_object( # pylint: disable=too-many-arguments,too-many-positional-arguments
self,
file_like: IO[Any],
remote_path: str,
file_size: int = 0,
callback: Optional[Callable[[int, int], Any]] = None,
confirm: bool = False,
) -> SFTPAttributes:
"""
Uploads a file-like object to the remote server.
:param file_like: File-like object to upload.
:type file_like: IO[Any]
:param remote_path: Remote path where object will be stored.
:type remote_path: str
:param file_size: Size of the file-like object (default: 0).
:type file_size: int
:param callback: Optional callback for progress monitoring.
:type callback: Optional[Callable[[int, int], Any]]
:param confirm: Whether to confirm the upload.
:type confirm: bool
:return: File attributes of the uploaded object.
:rtype: SFTPAttributes
:raises SftpClientError: If upload fails.
"""
try:
return self.client.putfo(
file_like,
remote_path,
file_size=file_size,
callback=callback,
confirm=confirm,
)
except IOError as error:
raise SftpClientError(f"Error uploading object: {error}") from error
[docs]
def delete(self, remote_path: str, is_folder: bool = False) -> None:
"""
Deletes a file or directory on the remote server.
:param remote_path: Path to the remote file or directory.
:type remote_path: str
:param is_folder: Whether the target is a folder (default: False).
:type is_folder: bool
:raises SftpClientError: If deletion fails.
"""
try:
if is_folder:
self.client.rmdir(remote_path)
else:
self.client.remove(remote_path)
except IOError as error:
raise SftpClientError(
f"Error deleting {'directory' if is_folder else 'file'}: {error}"
) from error
[docs]
class SftpClientError(Exception):
"""
Custom exception for SFTP operations.
Raised when SFTP operations fail due to connection issues,
authentication failures, or file system errors.
"""