This is an automated email from the ASF dual-hosted git repository.
shahar1 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 565515c18c9 Fix GCSToGCSOperator ignoring replace=False for
single-file copy (#66014)
565515c18c9 is described below
commit 565515c18c910a87b9efb3371f87e6e289e80ea0
Author: Taichi Nakashima (@taichi) <[email protected]>
AuthorDate: Wed May 6 01:45:37 2026 +0900
Fix GCSToGCSOperator ignoring replace=False for single-file copy (#66014)
---
.../providers/google/cloud/transfers/gcs_to_gcs.py | 19 +++++--
.../unit/google/cloud/transfers/test_gcs_to_gcs.py | 62 ++++++++++++++++++++++
2 files changed, 76 insertions(+), 5 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 823fef56021..12a75a5256c 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
@@ -416,11 +416,20 @@ class GCSToGCSOperator(BaseOperator):
# and copy directly
if len(objects) == 0 and prefix:
if hook.exists(self.source_bucket, prefix):
- uri = self._copy_single_object(
- hook=hook, source_object=prefix,
destination_object=self.destination_object
- )
- if uri:
- result_uris.append(uri)
+ # `objects` may have been emptied by _ignore_existing_files;
respect replace=False here too.
+ destination_object = self.destination_object or prefix
+ if not self.replace and hook.exists(self.destination_bucket,
destination_object):
+ self.log.info(
+ "Skipped object %s; already exists in destination
bucket %s.",
+ destination_object,
+ self.destination_bucket,
+ )
+ else:
+ uri = self._copy_single_object(
+ hook=hook, source_object=prefix,
destination_object=self.destination_object
+ )
+ if uri:
+ result_uris.append(uri)
elif self.source_object_required:
msg = f"{prefix} does not exist in bucket {self.source_bucket}"
self.log.warning(msg)
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 45dcff4516b..1c1f4268a1b 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
@@ -677,6 +677,68 @@ class TestGoogleCloudStorageToCloudStorageOperator:
):
operator.execute(None)
+ @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+ def
test_execute_replace_false_skips_single_file_when_destination_exists(self,
mock_hook):
+ mock_hook.return_value.list.side_effect = [
+ [SOURCE_OBJECT_NO_WILDCARD],
+ [SOURCE_OBJECT_NO_WILDCARD],
+ ]
+ mock_hook.return_value.exists.return_value = True
+
+ operator = GCSToGCSOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_object=SOURCE_OBJECT_NO_WILDCARD,
+ destination_bucket=DESTINATION_BUCKET,
+ destination_object=SOURCE_OBJECT_NO_WILDCARD,
+ replace=False,
+ )
+
+ operator.execute(None)
+
+ mock_hook.return_value.exists.assert_any_call(DESTINATION_BUCKET,
SOURCE_OBJECT_NO_WILDCARD)
+ mock_hook.return_value.rewrite.assert_not_called()
+
+ @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+ def
test_execute_replace_false_copies_single_file_when_destination_missing(self,
mock_hook):
+ mock_hook.return_value.list.side_effect = [[], []]
+ mock_hook.return_value.exists.side_effect = lambda bucket, _obj:
bucket == TEST_BUCKET
+
+ operator = GCSToGCSOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_object=SOURCE_OBJECT_NO_WILDCARD,
+ destination_bucket=DESTINATION_BUCKET,
+ destination_object=SOURCE_OBJECT_NO_WILDCARD,
+ replace=False,
+ )
+
+ operator.execute(None)
+
+ mock_hook.return_value.rewrite.assert_called_once_with(
+ TEST_BUCKET, SOURCE_OBJECT_NO_WILDCARD, DESTINATION_BUCKET,
SOURCE_OBJECT_NO_WILDCARD
+ )
+
+ @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+ def
test_execute_replace_true_copies_single_file_even_when_destination_exists(self,
mock_hook):
+ mock_hook.return_value.list.return_value = []
+ mock_hook.return_value.exists.return_value = True
+
+ operator = GCSToGCSOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_object=SOURCE_OBJECT_NO_WILDCARD,
+ destination_bucket=DESTINATION_BUCKET,
+ destination_object=SOURCE_OBJECT_NO_WILDCARD,
+ replace=True,
+ )
+
+ operator.execute(None)
+
+ mock_hook.return_value.rewrite.assert_called_once_with(
+ TEST_BUCKET, SOURCE_OBJECT_NO_WILDCARD, DESTINATION_BUCKET,
SOURCE_OBJECT_NO_WILDCARD
+ )
+
@pytest.mark.parametrize(
(
"existing_objects",
