dabla commented on code in PR #72336:
URL: https://github.com/apache/airflow/pull/72336#discussion_r4034478207
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -735,6 +741,74 @@ def get_files_by_pattern(self, path, fnmatch_pattern) ->
list[str]:
return matched_files
+ def transfer(
+ self,
+ operation: str,
+ local_filepath: str | list[str] | None,
+ remote_filepath: str | list[str],
+ confirm: bool = True,
+ create_intermediate_dirs: bool = False,
+ concurrency: int = 1,
+ prefetch: bool = True,
+ ) -> None:
+ """
+ Perform a synchronous SFTP transfer operation (GET, PUT, or DELETE).
+
+ Centralizes transfer logic so both the operator and the trigger
+ can delegate to the hook, in line with the DRY principle.
+
+ :param operation: The SFTP operation - put, get, or delete.
+ :param local_filepath: Local file path(s).
+ :param remote_filepath: Remote file path(s).
+ :param confirm: Whether to confirm file size after PUT (default: True).
+ :param create_intermediate_dirs: Create missing intermediate
directories (default: False).
+ :param concurrency: Number of threads for directory transfers
(default: 1).
+ :param prefetch: Whether to prefetch during GET (default: True).
+ """
+ if isinstance(local_filepath, str):
+ local_filepath_array = [local_filepath] if local_filepath else []
+ else:
+ local_filepath_array = local_filepath or []
+
+ if isinstance(remote_filepath, str):
+ remote_filepath_array = [remote_filepath]
+ else:
+ remote_filepath_array = list(remote_filepath)
+
+ if operation.lower() == SFTPOperation.GET:
+ for local, remote in zip(local_filepath_array,
remote_filepath_array):
+ if create_intermediate_dirs:
+ Path(os.path.dirname(local)).mkdir(parents=True,
exist_ok=True)
+ if self.isdir(remote):
+ if concurrency > 1:
+ self.retrieve_directory_concurrently(
+ remote, local, workers=concurrency,
prefetch=prefetch
+ )
+ else:
+ self.retrieve_directory(remote, local)
+ else:
+ self.retrieve_file(remote, local, prefetch=prefetch)
+ elif operation.lower() == SFTPOperation.PUT:
+ for local, remote in zip(local_filepath_array,
remote_filepath_array):
+ if create_intermediate_dirs:
+ self.create_directory(os.path.dirname(remote))
+ if os.path.isdir(local):
+ if concurrency > 1:
+ self.store_directory_concurrently(remote, local,
confirm=confirm, workers=concurrency)
+ else:
+ self.store_directory(remote, local, confirm=confirm)
+ else:
+ self.store_file(remote, local, confirm=confirm)
+ elif operation.lower() == SFTPOperation.DELETE:
+ for remote in remote_filepath_array:
+ if self.isdir(remote):
+ self.delete_directory(remote, include_files=True)
+ else:
+ try:
+ self.delete_file(remote)
+ except FileNotFoundError:
Review Comment:
Confirmed, the narrowing is intentional. paramiko reports a missing path
with `IOError(errno.ENOENT, text)` in `SFTPClient._convert_status`, and
Python's `OSError` constructor turns a two-argument `ENOENT` call into
`FileNotFoundError`, so the first two branches of the old helper were the same
case. The third branch only matched a single-argument `OSError(ENOENT)`, which
has `errno` set to `None`, and nothing in paramiko constructs that. The
original report in #56190 was a DELETE of a folder that does not exist yet; on
this branch `isdir()` returns `False` for it, `delete_file()` raises
`FileNotFoundError`, and the operator warns and skips as before. The async path
catches `asyncssh.SFTPNoSuchFile`, the equivalent there.
`test_delete_missing_file_warns` is now parametrized over
`FileNotFoundError("missing")` and `OSError(errno.ENOENT, "No such file")` in
14aea77d86, so the decision is locked in, and
`test_delete_permission_error_raises` still shows other `OSError`s propagate.
---
Drafted-by: Claude Code (Claude Fable 5.1); reviewed by @dabla before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
