Clients#

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")
class core_ftp.clients.sftp.SftpConnectionConfig(host: str, port: int = 22, user: str | None = None, password: str | None = None, private_key_path: str = '', passphrase: str | None = None)[source]#

Bases: object

Configuration for establishing an SFTP connection.

host: str#
port: int = 22#
user: str | None = None#
password: str | None = None#
private_key_path: str = ''#
passphrase: str | None = None#
__init__(host: str, port: int = 22, user: str | None = None, password: str | None = None, private_key_path: str = '', passphrase: str | None = None) → None#
class core_ftp.clients.sftp.SftpTransportConfig(transport_kwargs: Dict[str, Any] | None = None, connection_kwargs: Dict[str, Any] | None = None, disabled_algorithms: bool = False, algorithms_to_disable: List[str] | None = None)[source]#

Bases: object

Low-level Paramiko transport/connection options.

transport_kwargs: Dict[str, Any] | None = None#
connection_kwargs: Dict[str, Any] | None = None#
disabled_algorithms: bool = False#
algorithms_to_disable: List[str] | None = None#
__init__(transport_kwargs: Dict[str, Any] | None = None, connection_kwargs: Dict[str, Any] | None = None, disabled_algorithms: bool = False, algorithms_to_disable: List[str] | None = None) → None#
class core_ftp.clients.sftp.SftpClient(connection: SftpConnectionConfig, transport: SftpTransportConfig | None = None)[source]#

Bases: object

High-level SFTP client that wraps Paramiko for secure file transfer operations.

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")
__init__(connection: SftpConnectionConfig, transport: SftpTransportConfig | None = None) → None[source]#

Initialize SFTP client with connection parameters.

Parameters:
_sftp_client: SFTPClient | None#
_transport: Transport | None#
property client: SFTPClient#

Provides access to the underlying Paramiko SFTP client. Auto-connects if not already connected.

Returns:

The underlying SFTPClient instance.

Return type:

SFTPClient

Raises:

SftpClientError – If connection fails.

_ensure_transport() → Transport[source]#

Ensures transport connection exists, creating it if necessary.

Returns:

The transport instance.

Return type:

Transport.

connect() → Self[source]#

Establishes SFTP connection to the remote server.

Returns:

The SFTP client instance for method chaining.

Return type:

Self

Raises:

SftpClientError – If connection fails due to authentication, host key, SSH, or other errors.

close() → None[source]#

Closes the SFTP connection and transport. Safe to call multiple times or on unopened connections.

get_cwd() → str | None[source]#

Returns the current working directory on the remote server.

Returns:

Current working directory path, or None if not set.

Return type:

Optional[str]

Raises:

SftpClientError – If unable to get current directory.

chdir(remote_path: str) → None[source]#

Changes the current working directory on the remote server.

Parameters:

remote_path (str) – Path to change to.

Raises:

SftpClientError – If unable to change directory.

list_files(remote_path: str) → Iterator[Tuple[str, SFTPAttributes]][source]#

Read files under a remote directory.

Parameters:

remote_path – Remote directory path.

Returns:

Iterator of tuples in the form (“file_name”, SFTPAttributes)

download_file(remote_file_path: str, local_file_path: str) → str[source]#

Downloads a file from the remote server to local filesystem.

Parameters:

  • remote_file_path (str) – Path to the remote file.

  • local_file_path (str) – Local path where file will be saved.

Returns:

The local file path where file was saved.

Return type:

str

Raises:

SftpClientError – If download fails.

upload_file(file_path: str, remote_path: str, callback: Callable[[int, int], Any] | None = None, confirm: bool = False) → SFTPAttributes[source]#

Uploads a local file to the remote server.

Parameters:

  • file_path (str) – Local path to the file to upload.

  • remote_path (str) – Remote path where file will be stored.

  • callback (Optional[Callable[[int, int], Any]]) – Optional callback for progress monitoring.

  • confirm (bool) – Whether to confirm the upload.

Returns:

File attributes of the uploaded file.

Return type:

SFTPAttributes

Raises:

SftpClientError – If upload fails.

upload_object(file_like: IO[Any], remote_path: str, file_size: int = 0, callback: Callable[[int, int], Any] | None = None, confirm: bool = False) → SFTPAttributes[source]#

Uploads a file-like object to the remote server.

Parameters:

  • file_like (IO[Any]) – File-like object to upload.

  • remote_path (str) – Remote path where object will be stored.

  • file_size (int) – Size of the file-like object (default: 0).

  • callback (Optional[Callable[[int, int], Any]]) – Optional callback for progress monitoring.

  • confirm (bool) – Whether to confirm the upload.

Returns:

File attributes of the uploaded object.

Return type:

SFTPAttributes

Raises:

SftpClientError – If upload fails.

delete(remote_path: str, is_folder: bool = False) → None[source]#

Deletes a file or directory on the remote server.

Parameters:

  • remote_path (str) – Path to the remote file or directory.

  • is_folder (bool) – Whether the target is a folder (default: False).

Raises:

SftpClientError – If deletion fails.

exception core_ftp.clients.sftp.SftpClientError[source]#

Bases: Exception

Custom exception for SFTP operations.

Raised when SFTP operations fail due to connection issues, authentication failures, or file system errors.