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 the system OpenSSH client (the sftp binary driven in batch mode via os_helper.system) rather than an in-process SSH library. This is a deliberate choice: OpenSSH is the reference SSH implementation, ships on macOS, Linux and Windows 10+/Server 2019+, and — crucially — authenticates exactly the way the operator’s own ssh / sftp commands do. In particular it honours the SSH agent and lets sftp_key point at your public key (the recommended value: ~/.ssh/id_ed25519.pub — the agent / a hardware token then performs the signature, so no private-key material is named in the config) as readily as at a private key, which an in-process library cannot do.

Host key verification is on by default and cannot be disabled: every invocation passes StrictHostKeyChecking=yes, so a host whose key is not already in ~/.ssh/known_hosts is rejected. A caller who wants to trust an additional store may point cred["sftp_known_hosts"] at an extra known_hosts file.

The command is identical on every OS, so there is no per-platform branching: the only cross-platform concern is whether the sftp binary is installed, which is checked once, up front, with a clear error message.

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.

Only sftp_host, sftp_login and sftp_https are mandatory. Authentication (sftp_passwd / sftp_key) and the write location (sftp_destination_path, default "/") are optional and fall back to documented defaults — so a key-based login writing to the server root needs just the three required fields.

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 the three required keys always present, plus whichever optional keys were supplied. sftp_destination_path is always set (defaulting to "/").
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.

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='', *, retries=3, sha256=None, overwrite=True, resume=True, progress=True)[source]

Download a remote SFTP file (or directory) to local_path: resumable, retried, atomic, verified.

If sftp_address is a remote directory, every file under it is downloaded (recursively) via download_many() instead — see _download_folder(). sha256 is meaningless there (there is no single file to verify) and must be left None; the return value is local_path itself rather than a single file’s local path.

Mirrors os_helper.download_file’s design exactly, on the SFTP side (single-file case): bytes land in a <local_path>.part sidecar via reget (OpenSSH’s resume-download command), retried with exponential backoff — each retry continuing rather than restarting — and only os.replace-d onto local_path once size-verified (and, if sha256 is given, content-verified too). See _download_resumable() for the full design.

Parameters:

sftp_address (str) – Full sftp:// address or a plain remote path (file or directory) to fetch.
cred (dict) – Credentials dict.
local_path (str, optional) – Destination on the local disk. Defaults to the remote basename.
retries (int, optional) – Extra attempts on a failed or size-mismatched transfer, with exponential backoff (default 3, so up to four tries total). Each retry resumes rather than restarts.
sha256 (str or None, optional) – Expected lowercase hex SHA-256. When given, the finished file is verified and a mismatch raises ValueError (the bad sidecar is discarded). Only valid for a single-file download.
overwrite (bool, optional) – Re-download even if local_path already exists (default True, matching this function’s historical behaviour). Set False to reuse a complete destination without re-downloading — its digest is re-checked when sha256 is given. Ignored for a folder download.
resume (bool, optional) – Continue a partial .part sidecar (default True). Set False to discard any existing partial and download from scratch.
progress (bool, optional) – Show a bar on an interactive terminal (default True). Set False to suppress it even on a TTY (e.g. to keep a bulk loop’s output to one summary line instead of N bars).

Returns:

The local path of the downloaded file, or local_path itself for a folder download.

Return type:

str

Raises:

ValueError – If sha256 is given and the finished file’s digest does not match, or if sha256 is given for a directory sftp_address.
Exception – If every attempt fails. On total failure the .part sidecar is left in place so a later call to download() for the same destination resumes instead of starting over.

sftp_helper.main.download_many(files, cred, *, retries=3, archive=None, overwrite=True, resume=True, progress=True)[source]

Download several files in one bulk operation, archive-accelerated when possible.

The download-side mirror of upload_many(): on a server that also accepts exec, this collapses N downloads into one remote stage+zip, one get, and one local unzip — avoiding even the reduced per-file overhead ControlMaster reuse leaves behind (one get conversation per file otherwise). _probe_exec() detects that capability up front; a server without it (most SFTP-only hosting accounts) falls back to the ordinary per-file download() loop, unaffected, since ControlMaster reuse still applies there.

Parameters:

files (list of (str, str)) – (sftp_address, local_path) pairs to download — mirrors download()’s own argument order.
cred (dict) – Credentials dict.
retries (int, optional) – Forwarded to download() for the per-file fallback path (the archive path has its own stage+zip+get+unzip flow — a failure there falls all the way back to per-file, which does retry).
archive (bool or None, optional) – Force the archive path (True), force the per-file path (False), or auto-detect via _probe_exec() (None, default). Ignored (treated as False) when overwrite is False — see below.
overwrite (bool, optional) – Re-download every file unconditionally (default True). Set False to skip files already present locally — turning a repeat call into an incremental sync. This forces the per-file path regardless of archive: the archive path downloads and extracts the whole batch in one shot with no per-entry check, so it has no way to honour a per-file skip.
resume (bool) – Forwarded to download() for the per-file fallback path.
progress (bool) – Forwarded to download() for the per-file fallback path.

Returns:

{sftp_address: local_path} for every file.

Return type:

dict of str to str

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

Deprecated compatibility shim — validate the connection and yield cred.

The module used to hand back a live paramiko.SFTPClient. Now every operation runs as its own short-lived sftp batch, so there is no persistent client object to expose. This context manager is kept only so older with get_client_sftp(cred) as ...: call sites keep importing; it performs a cheap connectivity check (an ls of the server root, which also validates auth and the host key) and yields the credentials dict.

Yields:: dict – The credentials dict, after a successful connection has been proven.
Raises:: Exception – If the server cannot be reached or authentication fails.
Parameters:: cred (dict)
Return type:: Iterator[dict]

sftp_helper.main.list_dir(ftp_dir, cred, *, recursive=False)[source]

List a remote directory’s contents.

The missing counterpart to upload() for tooling that needs to know what is already on the server before deciding what to transfer (e.g. a mirror-style sync that skips or removes files, rather than upload()’s unconditional overwrite).

Parameters:

ftp_dir (str) – Full sftp:// address or a plain remote directory path.
cred (dict) – Credentials dict.
recursive (bool, optional) – When True, walks sub-directories too and returns paths relative to ftp_dir (e.g. "css/app.css"). When False (default), returns only the immediate entries (one path component each).

Returns:

Entry names/relative paths, in the order the server returned them.

Return type:

list of str

Raises:

Exception – Wrapped with the address if the connection fails or ftp_dir does not exist.

Examples

>>> import sftp_helper as sftph
>>> cred = sftph.credentials("settings.yaml")
>>> sftph.list_dir("/uploads", cred)
['2026-06', 'readme.txt']
>>> sftph.list_dir("/uploads", cred, recursive=True)
['readme.txt', '2026-06/report.pdf']

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

Recursively stat every file under a remote directory, in one walk.

The bulk counterpart to stat(): builds a full {relative_path: {"size", "mtime"}} map for an entire remote tree in one pass (one round trip per directory level, not per file), so a sync tool can decide what to upload by comparing against local files entirely in memory — no per-file remote round trip needed.

Parameters:

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

Returns:

{relative_posix_path: {"size": int, "mtime": datetime}}, one entry per file found anywhere under ftp_dir (directories are walked into, not listed). Empty if ftp_dir has no files.

Return type:

dict of str to dict

Raises:

Exception – Wrapped with the address if the connection fails or ftp_dir does not exist.

Examples

>>> import sftp_helper as sftph
>>> cred = sftph.credentials("settings.yaml")
>>> sftph.list_dir_stat("/uploads", cred)
{'readme.txt': {'size': 1234, 'mtime': ...}, '2026-06/report.pdf': {'size': 98765, 'mtime': ...}}

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.

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'

Raises:: ValueError – If path contains a newline, carriage return, NUL byte, or double quote — see _assert_safe_path(). Every remote-path-accepting function in this module funnels through here (directly or via strip_sftp_path()), so this is the one choke point that keeps such a path from ever reaching an sftp -b batch command or a remote ssh exec string, where it could break out of the quoting.
Parameters:: path (str)
Return type:: str

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.

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.

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_stat(sftp_address, cred)[source]

Return {"size": int, "mtime": datetime} for a remote file, or None.

Parses the server’s ls -l output (see _parse_ls_long_line()); mtime has minute resolution and no timezone (the server sends none over this interface), so treat it as an approximate local-clock reading good enough for “is the source newer” comparisons, not for anything requiring second-level or timezone-aware precision.

Parameters:

sftp_address (str) – Full sftp:// address or a plain remote path to a file (not a directory).
cred (dict) – Credentials dict.

Returns:

{"size": int, "mtime": datetime}, or None if the path does not exist (or is a directory — this is a file stat, not list_dir).

Return type:

dict or None

Raises:

Exception – Wrapped with the address if the connection fails.

Examples

>>> import sftp_helper as sftph
>>> cred = sftph.credentials("settings.yaml")
>>> sftph.remote_stat("/uploads/readme.txt", cred)
{'size': 1234, 'mtime': datetime.datetime(2026, 6, 19, 10, 30)}

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.
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='', *, retries=3, overwrite=True, resume=True, progress=True)[source]

Upload a local file (or directory) to the SFTP server: resumable, retried, atomic, size-verified.

If local_path is a directory, every non-hidden file under it is uploaded (recursively) via upload_many() instead — see _upload_folder(). sftp_address is then required (there is no single file to derive a content-hashed default from) and names the destination directory; the return value is sftp_address itself rather than a single file’s address.

If sftp_address is empty (single-file case), a content-hashed name under cred['sftp_destination_path'] is used.

Mirrors os_helper.download_file’s smart-transfer design: the file is sent into a temp remote path via reput (OpenSSH’s resume-upload command), retried with exponential backoff on failure — each retry continuing rather than restarting — and only published (renamed) onto sftp_address once its remote size matches the local file exactly. A caller only ever observes sftp_address absent or complete, never truncated. See _upload_resumable() for the full design (including its one honest gap versus download_file: size is verified, not a full content hash — that would need a remote exec channel many SFTP-only accounts don’t expose).

Parameters:

local_path (str) – Path to the local file or directory to upload.
cred (dict) – Credentials dict.
sftp_address (str, optional) – Destination address (or destination directory, for a folder upload). When empty for a single file, a deterministic content-hashed name is generated so identical files map to the same remote path.
retries (int, optional) – Extra attempts on a failed or size-mismatched transfer, with exponential backoff (default 3, so up to four tries total). Each retry resumes rather than restarts.
overwrite (bool, optional) – Re-upload even if sftp_address already exists (default True, matching download()’s own default). Set False to reuse an already-present destination without re-transferring: when sftp_address was auto-derived from the file’s content hash, mere presence already proves the right bytes are there (no further check needed — see the derivation above); when it was given explicitly, the skip only applies if the remote size already matches (upload has no cheap way to content-verify a remote file — see _upload_resumable()).
resume (bool, optional) – Continue a partial temp remote file (default True). Set False to discard any existing partial and upload from scratch.
progress (bool, optional) – Show a bar on an interactive terminal (default True). Set False to suppress it even on a TTY (e.g. to keep a bulk loop’s output to one summary line instead of N bars).

Returns:

The full sftp:// address (or plain remote path) of the file, or sftp_address itself for a folder upload.

Return type:

str

Raises:

ValueError – If local_path is a directory and sftp_address is empty.
Exception – If every attempt fails, or the atomic publish fails. On total failure the file is left in place at its temp remote path — see _UPLOAD_TMP_SUFFIX — so a later call to upload() for the same destination resumes instead of starting over.

sftp_helper.main.upload_many(files, cred, *, retries=3, archive=None, overwrite=True, resume=True, progress=True)[source]

Upload several files in one bulk operation, archive-accelerated when possible.

Every SFTP command in this package pays a fresh SSH handshake unless ControlMaster reuse kicks in (see _control_path()); on a server that also accepts exec, this collapses N file transfers into one zip upload plus one remote unzip — the biggest possible win, since it avoids even the reduced per-file overhead ControlMaster leaves behind (one put conversation, one directory-creation round trip, one publish-rename per file). _probe_exec() detects that capability up front; a server that does not have it (most SFTP-only hosting accounts) falls back to the ordinary per-file upload() loop, unaffected, since ControlMaster reuse still applies there.

Parameters:

files (list of (str, str)) – (local_path, sftp_address) pairs to upload.
cred (dict) – Credentials dict.
retries (int, optional) – Forwarded to upload() for the per-file fallback path (the archive path has its own upload+unzip retry-free flow — a failure there falls all the way back to per-file, which does retry).
archive (bool or None, optional) – Force the archive path (True), force the per-file path (False), or auto-detect via _probe_exec() (None, default). Ignored (treated as False) when overwrite is False — see below.
overwrite (bool, optional) – Re-upload every file unconditionally (default True). Set False to skip files already present with a matching size — turning a repeat call into an incremental sync. This forces the per-file path regardless of archive: the archive path zips and unzips the whole batch in one shot with no per-entry stat, so it has no way to honour a per-file skip.
resume (bool) – Forwarded to upload() for the per-file fallback path.
progress (bool) – Forwarded to upload() for the per-file fallback path.

Returns:

{local_path: sftp_address} for every file.

Return type:

dict of str to str