sftp_helper package

Submodules

• sftp_helper.api module
  • Credentials
  • Usage Example
  • Author
  • delete_endpoint()
  • dir_exists_endpoint()
  • download_endpoint()
  • exists_endpoint()
  • health()
  • mkdir_endpoint()
  • normalize_path_endpoint()
  • show_credentials()
  • strip_path_endpoint()
  • tempfile_endpoint()
  • upload_endpoint()
• sftp_helper.cli_argparse module
  • Subcommands
  • Usage Example
  • Author
  • build_parser()
  • main()
• sftp_helper.cli_click module
  • Design notes
  • Usage Example
  • Author
• sftp_helper.main module
  • credentials()
  • delete()
  • download()
  • get_client_sftp()
  • make_remote_directory()
  • normalize_path()
  • remote_dir_exist()
  • remote_file_exists()
  • remote_tempfile()
  • strip_sftp_path()
  • upload()
• sftp_helper.mcp module
  • Usage Example
  • Author
  • main()

Module contents

SFTP Helper — public API surface.

Re-exports the utility functions from sftp_helper.main so that downstream code can simply write import sftp_helper as sftph and reach every supported operation (credentials loading, connection context manager, upload, download, exists, delete, mkdir -p, remote temp file with auto-cleanup) without knowing about the module layout.

Backed by paramiko with strict host-key verification. See the module docs for the full policy — there is no flag to disable verification.

Usage Example

>>> import sftp_helper as sftph
>>> cred = sftph.credentials("sftp_config.json")
>>> sftph.upload("local.txt", cred, "/remote/base/local.txt")
>>> assert sftph.remote_file_exists("/remote/base/local.txt", cred)
>>> sftph.download("/remote/base/local.txt", cred, "roundtrip.txt")
>>> sftph.delete("/remote/base/local.txt", cred)

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

sftp_helper.credentials(config_path=None)

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.delete(sftp_address, cred)

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.download(sftp_address, cred, local_path='')

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.get_client_sftp(cred)

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.make_remote_directory(ftp_directory, cred)

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.normalize_path(path)

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.remote_dir_exist(ftp_dir, cred)

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.remote_file_exists(sftp_address, cred)

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.remote_tempfile(cred, ext='', subdir='')

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.strip_sftp_path(sftp_address, cred)

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.upload(local_path, cred, sftp_address='')

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.