potiuk commented on code in PR #72336:
URL: https://github.com/apache/airflow/pull/72336#discussion_r4062531298
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -735,6 +757,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)
Review Comment:
Correcting myself from the last round: I called this pre-existing and said
to skip it, but `SFTPHook.transfer()` is new in this PR, so it is in scope.
`retrieve_directory` already takes `prefetch`, and both sibling branches here
pass it:
```suggestion
self.retrieve_directory(remote, local,
prefetch=prefetch)
```
--
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]
