(airflow) branch main updated: refactor: allow to edit the file listings before copy (#49766)

[sh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

shahar 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 c3a4e8eeb2f refactor: allow to edit the file listings before copy (#49766)
c3a4e8eeb2f is described below

commit c3a4e8eeb2f4719f29eaebca891486a10ee0cd95
Author: Alexis BRENON <al...@greenbids.ai>
AuthorDate: Mon May 12 20:23:10 2025 +0200

    refactor: allow to edit the file listings before copy (#49766)
    
    * refactor: allow to edit the file listings before copy
    
    * fix static check
---
 .../providers/google/cloud/transfers/s3_to_gcs.py        | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/s3_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/s3_to_gcs.py
index 18687f1eb93..ba11be3bd5b 100644
--- a/providers/google/src/airflow/providers/google/cloud/transfers/s3_to_gcs.py
+++ b/providers/google/src/airflow/providers/google/cloud/transfers/s3_to_gcs.py
@@ -181,21 +181,27 @@ class S3ToGCSOperator(S3ListOperator):
                 'The destination Google Cloud Storage path must end with a slash "/" or be empty.'
             )
 
-    def execute(self, context: Context):
-        self._check_inputs()
+    def _get_files(self, context: Context, gcs_hook: GCSHook) -> list[str]:
         # use the super method to list all the files in an S3 bucket/key
         s3_objects = super().execute(context)
 
+        if not self.replace:
+            s3_objects = self.exclude_existing_objects(s3_objects=s3_objects, gcs_hook=gcs_hook)
+
+        return s3_objects
+
+    def execute(self, context: Context):
+        self._check_inputs()
         gcs_hook = GCSHook(
             gcp_conn_id=self.gcp_conn_id,
             impersonation_chain=self.google_impersonation_chain,
         )
-        if not self.replace:
-            s3_objects = self.exclude_existing_objects(s3_objects=s3_objects, gcs_hook=gcs_hook)
-
         s3_hook = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+
+        s3_objects = self._get_files(context, gcs_hook)
         if not s3_objects:
             self.log.info("In sync, no files needed to be uploaded to Google Cloud Storage")
+
         elif self.deferrable:
             self.transfer_files_async(s3_objects, gcs_hook, s3_hook)
         else: