(return a warning instead of an error when file is already absent)

Author: Eason09053360

providers/sftp/src/airflow/providers/sftp/operators/sftp.py

@@ -19,6 +19,7 @@
 
 from __future__ import annotations
 
+import errno
 import os
 import socket
 from collections.abc import Sequence
@@ -206,10 +207,18 @@ def execute(self, context: Any) -> str | list[str] | None:
                 for _remote_filepath in remote_filepath_array:
                     file_msg = f"{_remote_filepath}"
                     self.log.info("Starting to delete %s", file_msg)
-                    if self.sftp_hook.isdir(_remote_filepath):
-                        self.sftp_hook.delete_directory(_remote_filepath, include_files=True)
-                    else:
-                        self.sftp_hook.delete_file(_remote_filepath)
+                    try:
+                        if self.sftp_hook.isdir(_remote_filepath):
+                            self.sftp_hook.delete_directory(_remote_filepath, include_files=True)
+                        else:
+                            self.sftp_hook.delete_file(_remote_filepath)
+                    except OSError as exc:
+                        if self._is_missing_path_error(exc):
+                            self.log.warning(
+                                "Remote path %s does not exist. Skipping delete.", _remote_filepath
+                            )
+                            continue
+                        raise
 
         except Exception as e:
             raise AirflowException(
@@ -218,6 +227,16 @@ def execute(self, context: Any) -> str | list[str] | None:
 
         return self.local_filepath
 
+    @staticmethod
+    def _is_missing_path_error(exc: Exception) -> bool:
+        if isinstance(exc, FileNotFoundError):
+            return True
+        if isinstance(exc, OSError) and exc.errno == errno.ENOENT:
+            return True
+        if exc.args and isinstance(exc.args[0], int) and exc.args[0] == errno.ENOENT:
+            return True
+        return False
+
     def get_openlineage_facets_on_start(self):
         """
         Return OpenLineage datasets.

providers/sftp/tests/unit/sftp/operators/test_sftp.py

@@ -18,6 +18,8 @@
 from __future__ import annotations
 
 import contextlib
+import errno
+import logging
 import os
 import socket
 from base64 import b64encode
@@ -566,6 +568,39 @@ def test_str_dirpaths_delete(self, mock_delete):
         args, _ = mock_delete.call_args_list[0]
         assert args == (remote_filepath,)
 
+    @mock.patch("airflow.providers.sftp.operators.sftp.SFTPHook.delete_file")
+    @mock.patch("airflow.providers.sftp.operators.sftp.SFTPHook.isdir")
+    def test_delete_missing_file_warns(self, mock_isdir, mock_delete, caplog):
+        mock_isdir.return_value = False
+        mock_delete.side_effect = FileNotFoundError("missing")
+        remote_filepath = "/tmp/missing"
+        sftp_op = SFTPOperator(
+            task_id="test_missing_file_delete_warns",
+            sftp_hook=self.sftp_hook,
+            remote_filepath=remote_filepath,
+            operation=SFTPOperation.DELETE,
+        )
+
+        with caplog.at_level(logging.WARNING):
+            sftp_op.execute(None)
+
+        assert "does not exist. Skipping delete." in caplog.text
+
+    @mock.patch("airflow.providers.sftp.operators.sftp.SFTPHook.delete_file")
+    @mock.patch("airflow.providers.sftp.operators.sftp.SFTPHook.isdir")
+    def test_delete_permission_error_raises(self, mock_isdir, mock_delete):
+        mock_isdir.return_value = False
+        mock_delete.side_effect = PermissionError(errno.EACCES, "denied")
+        remote_filepath = "/tmp/protected"
+
+        with pytest.raises(AirflowException):
+            SFTPOperator(
+                task_id="test_permission_error_delete_raises",
+                sftp_hook=self.sftp_hook,
+                remote_filepath=remote_filepath,
+                operation=SFTPOperation.DELETE,
+            ).execute(None)
+
     @mock.patch("airflow.providers.sftp.operators.sftp.SFTPHook.delete_file")
     def test_local_filepath_exists_error_delete(self, mock_delete):
         local_filepath = "/tmp"