sftp_helper.main module

SFTP Helper

This module provides functions to interact with an SFTP server, allowing users to perform file uploads, downloads, and deletions, as well as to check file existence remotely.

Backed by paramiko’s SFTPClient. Host key verification is on by default: ~/.ssh/known_hosts is loaded, and unknown hosts are rejected. A caller who wants to trust a specific server may pass an additional known_hosts file via cred["sftp_known_hosts"] – there is no flag to disable verification.

Author: - Warith HARCHAOUI (https://linkedin.com/in/warith-harchaoui)

sftp_helper.main.credentials(config_path=None)[source]

Retrieve SFTP credentials from a configuration file, folder, or environment.

Parameters:: config_path (str) – Path to a JSON/YAML file, a directory containing one, or None to fall back to environment variables / .env.
Returns:: Dictionary with keys: sftp_host, sftp_login, sftp_passwd, sftp_destination_path, sftp_https. sftp_port and sftp_known_hosts are optional.
Return type:: dict

sftp_helper.main.delete(sftp_address, cred)[source]

Delete a remote file. Returns True if the file is gone afterwards (including the case where it never existed).

Parameters:

sftp_address (str) – Full sftp:// address or a plain remote path.
cred (dict) – Credentials dict passed straight to get_client_sftp().

Returns:

Always True on success — deleting an absent file is a no-op, which makes the operation idempotent.

Return type:

bool

Raises:

Exception – Wrapped with the address if the connection or removal fails.

sftp_helper.main.download(sftp_address, cred, local_path='')[source]

Download a remote SFTP file to local_path (defaults to the remote basename).

Parameters:

sftp_address (str) – Full sftp:// address or a plain remote path to fetch.
cred (dict) – Credentials dict passed straight to get_client_sftp().
local_path (str, optional) – Destination on the local disk. Defaults to the remote basename.

Returns:

The local path of the downloaded file.

Return type:

str

Raises:

Exception – Wrapped with both paths if the connection or transfer fails.

sftp_helper.main.get_client_sftp(cred)[source]

Open an SFTP connection with strict host key verification.

The system ~/.ssh/known_hosts is loaded automatically. If cred["sftp_known_hosts"] is set, that file is loaded as well. Connecting to a host whose key is not in either store raises paramiko.SSHException – there is no opt-out.

Authentication tries password first (if sftp_passwd is non-empty), then falls back to the SSH agent and default identity files (~/.ssh/id_rsa, ~/.ssh/id_ed25519 …).

Yields:: paramiko.SFTPClient
Parameters:: cred (dict)
Return type:: Iterator[paramiko.SFTPClient]

sftp_helper.main.make_remote_directory(ftp_directory, cred)[source]

Ensure the specified remote directory exists, creating intermediate levels as needed.

Parameters:

ftp_directory (str) – Full sftp:// address or a plain remote directory path. Every missing intermediate level is created (mkdir -p semantics).
cred (dict) – Credentials dict passed straight to get_client_sftp().

Raises:

AssertionError – If the target directory is still absent after the create loop.

Return type:

None

sftp_helper.main.normalize_path(path)[source]

Normalize a remote path: ensure single leading ‘/’, strip trailing slashes.

Parameters:: path (str) – A raw remote path, possibly missing the leading slash or carrying redundant trailing slashes.
Returns:: The canonical form (single leading ‘/’, no trailing ‘/’); the root "/" is preserved rather than collapsed to the empty string.
Return type:: str

Examples

>>> normalize_path("foo/bar///")
'/foo/bar'

sftp_helper.main.remote_dir_exist(ftp_dir, cred)[source]

Return True iff the remote directory exists.

Parameters:

ftp_dir (str) – Full sftp:// address or a plain remote directory path.
cred (dict) – Credentials dict passed straight to get_client_sftp().

Returns:

Whether the remote path exists and is a directory.

Return type:

bool

sftp_helper.main.remote_file_exists(sftp_address, cred)[source]

Return True iff the remote path exists.

Parameters:

sftp_address (str) – Full sftp:// address or a plain remote path.
cred (dict) – Credentials dict passed straight to get_client_sftp().

Returns:

Whether the remote file exists.

Return type:

bool

Raises:

Exception – Wrapped with the address if the connection or probe fails.

sftp_helper.main.remote_tempfile(cred, ext='', subdir='')[source]

Reserve a unique remote path under cred['sftp_destination_path'] and delete it on exit.

Parameters:

cred (dict) – Credentials dict passed straight to get_client_sftp().
ext (str, optional) – File extension for the reserved name (with or without the leading dot).
subdir (str, optional) – Subdirectory under sftp_destination_path; created if missing.

Yields:

(sftp_address, https_url) – The reserved remote location – the file does not exist yet; the caller is expected to upload to it (or skip entirely, in which case cleanup is a no-op).
Cleanup
——-
The remote file is deleted in finally. Cleanup failures re-raise only
if no other exception is already propagating; otherwise they are logged
so the original error survives.

Return type:

Iterator[tuple[str, str]]

Example

>>> with remote_tempfile(cred, ext="txt") as (addr, url):
...     upload("local.txt", cred, addr)
...     assert osh.is_working_url(url)

sftp_helper.main.strip_sftp_path(sftp_address, cred)[source]

Strip sftp:// and the host from an SFTP address.

Idempotent: passing an already-stripped path returns it unchanged (modulo normalization).

Parameters:

sftp_address (str) – Either a full sftp://host/path address or a plain remote path.
cred (dict) – Credentials dict; only cred["sftp_host"] is read, to know which host token to remove.

Returns:

The normalized remote path with scheme and host removed.

Return type:

str

sftp_helper.main.upload(local_path, cred, sftp_address='')[source]

Upload a local file to the SFTP server.

If sftp_address is empty, a content-hashed name under cred['sftp_destination_path'] is used.

Parameters:

local_path (str) – Path to the local file to upload.
cred (dict) – Credentials dict passed straight to get_client_sftp().
sftp_address (str, optional) – Destination address. When empty, a deterministic content-hashed name is generated so identical files map to the same remote path.

Returns:

The full sftp:// address (or plain remote path) of the file.

Return type:

str

Raises:

Exception – Wrapped with both paths if the connection or transfer fails.