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 b3cbcb7d1d3 Return list of GCS URIs from SalesforceToGcsOperator
(#61659)
b3cbcb7d1d3 is described below
commit b3cbcb7d1d3f852520499f0f51f44534031ebbfa
Author: Pratiksha Badheka <[email protected]>
AuthorDate: Tue May 5 21:37:10 2026 +0530
Return list of GCS URIs from SalesforceToGcsOperator (#61659)
---
.../google/cloud/transfers/salesforce_to_gcs.py | 23 ++++++++++++++++++++--
.../cloud/transfers/test_salesforce_to_gcs.py | 14 +++++++++++--
2 files changed, 33 insertions(+), 4 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/salesforce_to_gcs.py
b/providers/google/src/airflow/providers/google/cloud/transfers/salesforce_to_gcs.py
index c2c42ec68c1..d736eacb563 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/salesforce_to_gcs.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/salesforce_to_gcs.py
@@ -18,6 +18,7 @@ from __future__ import annotations
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 @@ class SalesforceToGcsOperator(BaseOperator):
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 @@ class SalesforceToGcsOperator(BaseOperator):
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
unwrap_single=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 @@ class SalesforceToGcsOperator(BaseOperator):
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]
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_salesforce_to_gcs.py
b/providers/google/tests/unit/google/cloud/transfers/test_salesforce_to_gcs.py
index 85d056b30c9..1e9887c5824 100644
---
a/providers/google/tests/unit/google/cloud/transfers/test_salesforce_to_gcs.py
+++
b/providers/google/tests/unit/google/cloud/transfers/test_salesforce_to_gcs.py
@@ -45,13 +45,22 @@ SALESFORCE_RESPONSE = {
INCLUDE_DELETED = True
QUERY_PARAMS = {"DEFAULT_SETTING": "ENABLED"}
+pytestmark = pytest.mark.filterwarnings("ignore::FutureWarning")
+
[email protected](
+ ("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 @@ class TestSalesforceToGcsOperator:
coerce_to_timestamp=True,
record_time_added=True,
task_id=TASK_ID,
+ unwrap_single=unwrap_single,
)
result = operator.execute({})
@@ -85,4 +95,4 @@ class TestSalesforceToGcsOperator:
bucket_name=GCS_BUCKET, object_name=GCS_OBJECT_PATH,
filename=mock.ANY, gzip=False
)
- assert result == EXPECTED_GCS_URI
+ assert result == expected
