Return list of GCS URIs from SalesforceToGcsOperator

providers/google/src/airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

@@ -18,6 +18,7 @@
 
 import os
 import tempfile
+import warnings
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
@@ -52,6 +53,10 @@ class SalesforceToGcsOperator(BaseOperator):
         to the resulting data that marks when the data was fetched from Salesforce. Default: False
     :param gzip: Option to compress local file or file data for upload
     :param gcp_conn_id: the name of the connection that has the parameters we need to connect to GCS.
+    :param unwrap_single: If True, unwrap a single-element result list into a plain string value
+        for backward compatibility. If False, always return a list of GCS URIs.
+        If not explicitly provided, defaults to True and emits a FutureWarning that
+        the default will change to False in a future release.
     """
 
     template_fields: Sequence[str] = (
@@ -76,6 +81,7 @@ def __init__(
         record_time_added: bool = False,
         gzip: bool = False,
         gcp_conn_id: str = "google_cloud_default",
+        unwrap_single: bool | None = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -90,8 +96,18 @@ def __init__(
         self.gcp_conn_id = gcp_conn_id
         self.include_deleted = include_deleted
         self.query_params = query_params
+        if unwrap_single is None:
+            warnings.warn(
+                "Returning a list of Salesforce FileShare filenames from SalesforceToGcsOperator is deprecated and "
+                "will change to list[str] of GCS URIs in a future release. Set return_gcs_uris=True to opt in.",
+                FutureWarning,
+                stacklevel=2,
+            )
+            unwrap_single = True
+        else:
+            self.unwrap_single = unwrap_single
 
-    def execute(self, context: Context):
+    def execute(self, context: Context) -> str | list[str]:
         salesforce = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
         response = salesforce.make_query(
             query=self.query, include_deleted=self.include_deleted, query_params=self.query_params
@@ -117,4 +133,7 @@ def execute(self, context: Context):
 
             gcs_uri = f"gs://{self.bucket_name}/{self.object_name}"
             self.log.info("%s uploaded to GCS", gcs_uri)
-            return gcs_uri
+
+            if self.unwrap_single:
+                return gcs_uri
+            return [gcs_uri]

providers/google/tests/unit/google/cloud/transfers/test_salesforce_to_gcs.py

@@ -45,13 +45,22 @@
 INCLUDE_DELETED = True
 QUERY_PARAMS = {"DEFAULT_SETTING": "ENABLED"}
 
+pytestmark = pytest.mark.filterwarnings("ignore::FutureWarning")
 
+
+@pytest.mark.parametrize(
+    ("unwrap_single", "expected"),
+    [
+        (True, EXPECTED_GCS_URI),
+        (False, [EXPECTED_GCS_URI]),
+    ],
+)
 class TestSalesforceToGcsOperator:
     @pytest.mark.db_test
     @mock.patch.object(GCSHook, "upload")
     @mock.patch.object(SalesforceHook, "write_object_to_file")
     @mock.patch.object(SalesforceHook, "make_query")
-    def test_execute(self, mock_make_query, mock_write_object_to_file, mock_upload):
+    def test_execute(self, mock_make_query, mock_write_object_to_file, mock_upload, unwrap_single, expected):
         mock_make_query.return_value = SALESFORCE_RESPONSE
 
         operator = SalesforceToGcsOperator(
@@ -66,6 +75,7 @@ def test_execute(self, mock_make_query, mock_write_object_to_file, mock_upload):
             coerce_to_timestamp=True,
             record_time_added=True,
             task_id=TASK_ID,
+            unwrap_single=unwrap_single,
         )
         result = operator.execute({})
 
@@ -85,4 +95,4 @@ def test_execute(self, mock_make_query, mock_write_object_to_file, mock_upload):
             bucket_name=GCS_BUCKET, object_name=GCS_OBJECT_PATH, filename=mock.ANY, gzip=False
         )
 
-        assert result == EXPECTED_GCS_URI
+        assert result == expected