Repository: incubator-airflow Updated Branches: refs/heads/master 0c5ebcbd1 -> 78f718cb9
[AIRFLOW-2622] add confirm option to SFTPOperator[] surfaces the confirm option in the SFTPOperator provided by the underlying parmiko library, useful for when the receiving server moves the incoming file before the confirmation step can be completed Closes #3542 from caddac/master Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/78f718cb Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/78f718cb Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/78f718cb Branch: refs/heads/master Commit: 78f718cb99fae1aeb3f757d448ef336a6d6061ca Parents: 0c5ebcb Author: Dave Cavaletto <[email protected]> Authored: Fri Jul 6 22:36:03 2018 +0200 Committer: Fokko Driesprong <[email protected]> Committed: Fri Jul 6 22:36:03 2018 +0200 ---------------------------------------------------------------------- airflow/contrib/operators/sftp_operator.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/78f718cb/airflow/contrib/operators/sftp_operator.py ---------------------------------------------------------------------- diff --git a/airflow/contrib/operators/sftp_operator.py b/airflow/contrib/operators/sftp_operator.py index 153f440..cf6c6d2 100644 --- a/airflow/contrib/operators/sftp_operator.py +++ b/airflow/contrib/operators/sftp_operator.py @@ -43,8 +43,10 @@ class SFTPOperator(BaseOperator): :type local_filepath: str :param remote_filepath: remote file path to get or put. (templated) :type remote_filepath: str - :param operation: specify operation 'get' or 'put', defaults to get + :param operation: specify operation 'get' or 'put', defaults to put :type get: bool + :param confirm: specify if the SFTP operation should be confirmed, defaults to True + :type confirm: bool """ template_fields = ('local_filepath', 'remote_filepath') @@ -56,6 +58,7 @@ class SFTPOperator(BaseOperator): local_filepath=None, remote_filepath=None, operation=SFTPOperation.PUT, + confirm=True, *args, **kwargs): super(SFTPOperator, self).__init__(*args, **kwargs) @@ -65,6 +68,7 @@ class SFTPOperator(BaseOperator): self.local_filepath = local_filepath self.remote_filepath = remote_filepath self.operation = operation + self.confirm = confirm if not (self.operation.lower() == SFTPOperation.GET or self.operation.lower() == SFTPOperation.PUT): raise TypeError("unsupported operation value {0}, expected {1} or {2}" @@ -93,7 +97,9 @@ class SFTPOperator(BaseOperator): file_msg = "from {0} to {1}".format(self.local_filepath, self.remote_filepath) self.log.debug("Starting to transfer file %s", file_msg) - sftp_client.put(self.local_filepath, self.remote_filepath) + sftp_client.put(self.local_filepath, + self.remote_filepath, + confirm=self.confirm) except Exception as e: raise AirflowException("Error while transferring {0}, error: {1}"
