This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f87240f7031 Handling `NotFound` when file has already been deleted
from GCS (#72275)
f87240f7031 is described below
commit f87240f703181c3807d2cddc7f39cedbe3712a3a
Author: Jake McGrath <[email protected]>
AuthorDate: Sat Sep 12 02:56:42 2026 -0400
Handling `NotFound` when file has already been deleted from GCS (#72275)
* fix/issue-72264: Handling NotFound when file has already been deleted
* fix/issue-72264: Fixing failing tests for Python 3.10
* Apply batched suggestions from code review
Co-authored-by: Ryan Hatter <[email protected]>
* fix/issue-72264: Fixing variable mis-match
---------
Co-authored-by: Ryan Hatter <[email protected]>
---
.../providers/google/cloud/transfers/gcs_to_gcs.py | 13 +++++++--
.../unit/google/cloud/transfers/test_gcs_to_gcs.py | 34 ++++++++++++++++++++++
2 files changed, 45 insertions(+), 2 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
index b6f7e4f3729..8d10763754e 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
@@ -23,6 +23,8 @@ import warnings
from collections.abc import Sequence
from typing import TYPE_CHECKING
+from google.api_core.exceptions import NotFound
+
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.google.cloud.hooks.gcs import GCSHook
@@ -607,8 +609,15 @@ class GCSToGCSOperator(BaseOperator):
**rewrite_kwargs,
)
- if self.move_object:
- hook.delete(self.source_bucket, source_object)
+ try:
+ if self.move_object:
+ hook.delete(self.source_bucket, source_object)
+
+ # Handle case where the file has already been deleted and a NotFound
exception is raised
+ except NotFound:
+ self.log.warning(
+ "Object %s does not exist in the source bucket %s",
source_object, self.source_bucket
+ )
return f"gs://{dest_bucket}/{destination_object}"
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py
b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py
index bfd2c8f586f..73f185adead 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py
@@ -21,6 +21,7 @@ from datetime import datetime
from unittest import mock
import pytest
+from google.api_core.exceptions import NotFound
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.compat.openlineage.facet import Dataset
@@ -547,6 +548,39 @@ class TestGoogleCloudStorageToCloudStorageOperator:
any_order=True,
)
+ @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+ def test_executes_with_source_objects_not_found(self, mock_hook, caplog):
+ """Handle a file NotFound exception when attempting to delete source
object."""
+ operator = GCSToGCSOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_objects=SOURCE_OBJECTS_SINGLE_FILE,
+ move_object=True,
+ )
+
+ mock_hook.return_value.delete.side_effect = NotFound("Object not
found")
+
+ operator.execute(None)
+
+ expected_object = SOURCE_OBJECTS_SINGLE_FILE[0]
+ expected_message = f"Object {expected_object} does not exist in the
source bucket {TEST_BUCKET}"
+ assert any(expected_message in record.getMessage() for record in
caplog.records)
+
+ @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+ def test_executes_with_source_objects_failed_delete(self, mock_hook):
+ """Handle a non-NotFound exception when attempting to delete a
non-existent object."""
+ operator = GCSToGCSOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_objects=SOURCE_OBJECTS_SINGLE_FILE,
+ move_object=True,
+ )
+
+ mock_hook.return_value.delete.side_effect = RuntimeError
+
+ with pytest.raises(RuntimeError):
+ operator.execute(None)
+
@mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
def test_executes_with_a_delimiter(self, mock_hook):
operator = GCSToGCSOperator(
