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 b9c52951378 Create GCS folder markers as directories in
GCSToSFTPOperator (#73880)
b9c52951378 is described below
commit b9c529513784531304b859c9bcf37adb6ce1aa6b
Author: Yuseok Jo <[email protected]>
AuthorDate: Sat Oct 3 16:21:47 2026 +0900
Create GCS folder markers as directories in GCSToSFTPOperator (#73880)
---
providers/google/docs/changelog.rst | 9 +++
.../google/cloud/transfers/gcs_to_sftp.py | 55 ++++++++++++-----
.../google/cloud/transfers/test_gcs_to_sftp.py | 72 ++++++++++++++++++++++
3 files changed, 119 insertions(+), 17 deletions(-)
diff --git a/providers/google/docs/changelog.rst
b/providers/google/docs/changelog.rst
index be32f9edc07..e0b19f6049b 100644
--- a/providers/google/docs/changelog.rst
+++ b/providers/google/docs/changelog.rst
@@ -27,6 +27,15 @@
Changelog
---------
+.. warning::
+ ``GCSToSFTPOperator`` no longer writes GCS folder markers, the objects
whose name ends with
+ ``/``, as files. They were written as an empty file at the folder's path,
so the folder's
+ contents could not be transferred after them. With
``create_intermediate_dirs=True``, the
+ default, each marker is now created as a directory on the SFTP server, and
deleted from the
+ bucket when ``move_object=True``. With ``create_intermediate_dirs=False``,
markers are skipped
+ and left in the bucket. An empty file that an earlier version wrote at a
marker's path blocks
+ the directory and fails the task; remove it from the SFTP server before
the next run.
+
22.6.0
......
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
index 521125b610b..bc08bb6dc0b 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
@@ -82,7 +82,9 @@ class GCSToSFTPOperator(BaseOperator):
:param keep_directory_structure: (Optional) When set to False the path of
the file
on the bucket is recreated within path passed in destination_path.
:param create_intermediate_dirs: (Optional) When set to True the
intermediate directories
- in the specified file path will be created.
+ in the specified file path will be created. It also controls folder
markers (objects whose
+ name ends with ``/``): when True, each marker is created as a
directory on the SFTP server;
+ when False, it is skipped and left in the bucket, even if
``move_object`` is True.
:param move_object: When move object is True, the object is moved instead
of copied to the new location. This is the equivalent of a mv command
as opposed to a cp command.
@@ -209,25 +211,44 @@ class GCSToSFTPOperator(BaseOperator):
destination_path: str,
) -> None:
"""Copy single object."""
- self.log.info(
- "Executing copy of gs://%s/%s to %s",
- self.source_bucket,
- source_object,
- destination_path,
- )
+ # ``_resolve_destination_path`` strips the trailing slash, so storing
a folder marker
+ # would write a file at the path its children need as a directory.
+ if source_object.endswith("/"):
+ if not self.create_intermediate_dirs:
+ self.log.info(
+ "Skipping folder marker gs://%s/%s because
create_intermediate_dirs is False",
+ self.source_bucket,
+ source_object,
+ )
+ return
+
+ self.log.info(
+ "Creating directory %s for folder marker gs://%s/%s",
+ destination_path,
+ self.source_bucket,
+ source_object,
+ )
+ sftp_hook.create_directory(destination_path)
+ else:
+ self.log.info(
+ "Executing copy of gs://%s/%s to %s",
+ self.source_bucket,
+ source_object,
+ destination_path,
+ )
- dir_path = os.path.dirname(destination_path)
+ dir_path = os.path.dirname(destination_path)
- if self.create_intermediate_dirs:
- sftp_hook.create_directory(dir_path)
+ if self.create_intermediate_dirs:
+ sftp_hook.create_directory(dir_path)
- with NamedTemporaryFile("w") as tmp:
- gcs_hook.download(
- bucket_name=self.source_bucket,
- object_name=source_object,
- filename=tmp.name,
- )
- sftp_hook.store_file(destination_path, tmp.name)
+ with NamedTemporaryFile("w") as tmp:
+ gcs_hook.download(
+ bucket_name=self.source_bucket,
+ object_name=source_object,
+ filename=tmp.name,
+ )
+ sftp_hook.store_file(destination_path, tmp.name)
if self.move_object:
self.log.info("Executing delete of gs://%s/%s",
self.source_bucket, source_object)
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_sftp.py
b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_sftp.py
index 8bc93419338..7955d3c2d49 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_sftp.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_sftp.py
@@ -467,6 +467,78 @@ class TestGoogleCloudStorageToSFTPOperator:
sftp_hook_mock.return_value.create_directory.assert_not_called()
+ @pytest.mark.parametrize(
+ ("move_object", "expected_deleted"),
+ [
+ pytest.param(False, [], id="copy"),
+ pytest.param(True, ["data/empty/", "data/folder/",
"data/folder/file.txt"], id="move"),
+ ],
+ )
+
@mock.patch("airflow.providers.google.cloud.transfers.gcs_to_sftp.GCSHook",
autospec=True)
+
@mock.patch("airflow.providers.google.cloud.transfers.gcs_to_sftp.SFTPHook",
autospec=True)
+ def test_folder_markers_are_created_as_directories(
+ self, sftp_hook_mock, gcs_hook_mock, move_object, expected_deleted
+ ):
+ gcs_hook = gcs_hook_mock.return_value
+ sftp_hook = sftp_hook_mock.return_value
+ gcs_hook.list.return_value = ["data/empty/", "data/folder/",
"data/folder/file.txt"]
+ operator = GCSToSFTPOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_object="data/*",
+ destination_path=DESTINATION_SFTP,
+ create_intermediate_dirs=True,
+ move_object=move_object,
+ gcp_conn_id=GCP_CONN_ID,
+ sftp_conn_id=SFTP_CONN_ID,
+ )
+ operator.execute(None)
+
+
sftp_hook.create_directory.assert_any_call(os.path.join(DESTINATION_SFTP,
"data", "empty"))
+ sftp_hook.store_file.assert_called_once_with(
+ os.path.join(DESTINATION_SFTP, "data", "folder", "file.txt"),
mock.ANY
+ )
+ gcs_hook.download.assert_called_once_with(
+ bucket_name=TEST_BUCKET, object_name="data/folder/file.txt",
filename=mock.ANY
+ )
+ assert gcs_hook.delete.call_args_list == [mock.call(TEST_BUCKET, obj)
for obj in expected_deleted]
+
+ @pytest.mark.parametrize(
+ ("move_object", "expected_deleted"),
+ [
+ pytest.param(False, [], id="copy"),
+ pytest.param(True, ["data/folder/file.txt"], id="move"),
+ ],
+ )
+
@mock.patch("airflow.providers.google.cloud.transfers.gcs_to_sftp.GCSHook",
autospec=True)
+
@mock.patch("airflow.providers.google.cloud.transfers.gcs_to_sftp.SFTPHook",
autospec=True)
+ def test_folder_markers_are_skipped_without_intermediate_dirs(
+ self, sftp_hook_mock, gcs_hook_mock, move_object, expected_deleted
+ ):
+ gcs_hook = gcs_hook_mock.return_value
+ sftp_hook = sftp_hook_mock.return_value
+ gcs_hook.list.return_value = ["data/empty/", "data/folder/",
"data/folder/file.txt"]
+ operator = GCSToSFTPOperator(
+ task_id=TASK_ID,
+ source_bucket=TEST_BUCKET,
+ source_object="data/*",
+ destination_path=DESTINATION_SFTP,
+ create_intermediate_dirs=False,
+ move_object=move_object,
+ gcp_conn_id=GCP_CONN_ID,
+ sftp_conn_id=SFTP_CONN_ID,
+ )
+ operator.execute(None)
+
+ sftp_hook.create_directory.assert_not_called()
+ sftp_hook.store_file.assert_called_once_with(
+ os.path.join(DESTINATION_SFTP, "data", "folder", "file.txt"),
mock.ANY
+ )
+ gcs_hook.download.assert_called_once_with(
+ bucket_name=TEST_BUCKET, object_name="data/folder/file.txt",
filename=mock.ANY
+ )
+ assert gcs_hook.delete.call_args_list == [mock.call(TEST_BUCKET, obj)
for obj in expected_deleted]
+
@pytest.mark.parametrize(
"source_object",
[
