From 405c7b559e56f02c8d54ecf6e4e99b596b54a23e Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 25 Jan 2026 13:50:41 -0800 Subject: [PATCH 1/7] Refactor AzureBlobStorageToGCSOperator and AzureFileShareToGCSOperator to return lists of GCS URIs instead of strings --- .../providers/google/cloud/transfers/azure_blob_to_gcs.py | 4 ++-- .../google/cloud/transfers/azure_fileshare_to_gcs.py | 6 ++++-- .../unit/google/cloud/transfers/test_azure_blob_to_gcs.py | 3 ++- .../google/cloud/transfers/test_azure_fileshare_to_gcs.py | 3 ++- 4 files changed, 10 insertions(+), 6 deletions(-) diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py index 1ace104c28297..21d43d84b03a8 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py @@ -94,7 +94,7 @@ def __init__( "filename", ) - def execute(self, context: Context) -> str: + def execute(self, context: Context) -> list[str]: azure_hook = WasbHook(wasb_conn_id=self.wasb_conn_id) gcs_hook = GCSHook( gcp_conn_id=self.gcp_conn_id, @@ -122,7 +122,7 @@ def execute(self, context: Context) -> str: self.blob_name, self.bucket_name, ) - return f"gs://{self.bucket_name}/{self.object_name}" + return [f"gs://{self.bucket_name}/{self.object_name}"] def get_openlineage_facets_on_start(self): from airflow.providers.common.compat.openlineage.facet import Dataset diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py index ad362c508ba04..eea4a8347f0cf 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py @@ -125,7 +125,7 @@ def _check_inputs(self) -> None: 'The destination Google Cloud Storage path must end with a slash "/" or be empty.' ) - def execute(self, context: Context): + def execute(self, context: Context) -> list[str]: self._check_inputs() azure_fileshare_hook = AzureFileShareHook( share_name=self.share_name, @@ -162,6 +162,7 @@ def execute(self, context: Context): files = list(set(files) - set(existing_files)) + uploaded_gcs_uris: list[str] = [] if files: self.log.info("%s files are going to be synced.", len(files)) if self.directory_path is None: @@ -181,9 +182,10 @@ def execute(self, context: Context): # enforced at instantiation time dest_gcs_object = dest_gcs_object_prefix + file gcs_hook.upload(dest_gcs_bucket, dest_gcs_object, temp_file.name, gzip=self.gzip) + uploaded_gcs_uris.append(f"gs://{dest_gcs_bucket}/{dest_gcs_object}") self.log.info("All done, uploaded %d files to Google Cloud Storage.", len(files)) else: self.log.info("There are no new files to sync. Have a nice day!") self.log.info("In sync, no files needed to be uploaded to Google Cloud Storage") - return files + return uploaded_gcs_uris diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py index aa575eb9df40b..d51d05885d90d 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py @@ -74,7 +74,7 @@ def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb): task_id=TASK_ID, ) - op.execute(context=None) + result = op.execute(context=None) mock_hook_wasb.assert_called_once_with(wasb_conn_id=WASB_CONN_ID) mock_hook_wasb.return_value.get_file.assert_called_once_with( @@ -91,6 +91,7 @@ def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb): gzip=GZIP, filename=mock_temp.NamedTemporaryFile.return_value.__enter__.return_value.name, ) + assert result == [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"] @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.WasbHook") def test_execute_single_file_transfer_openlineage(self, mock_hook_wasb): diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py index 39b5bb62f3859..31e9cfcc3cfc1 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py @@ -87,7 +87,8 @@ def test_execute(self, gcs_mock_hook, azure_fileshare_mock_hook): impersonation_chain=IMPERSONATION_CHAIN, ) - assert sorted(MOCK_FILES) == sorted(uploaded_files) + expected_files = [f"gs://gcs-bucket/data/{file_name}" for file_name in MOCK_FILES] + assert sorted(expected_files) == sorted(uploaded_files) @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook") From 037243b7e31f093cd27def3561bd43bb8cab910e Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 25 Jan 2026 15:39:05 -0800 Subject: [PATCH 2/7] Add `return_gcs_uris` parameter and deprecation warnings for legacy returns --- .../cloud/transfers/azure_blob_to_gcs.py | 19 +++++++++++++++++-- .../cloud/transfers/azure_fileshare_to_gcs.py | 15 ++++++++++++++- .../cloud/transfers/test_azure_blob_to_gcs.py | 16 ++++++++++++++-- .../transfers/test_azure_fileshare_to_gcs.py | 15 +++++++++++++-- 4 files changed, 58 insertions(+), 7 deletions(-) diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py index 21d43d84b03a8..a34ceed789f3f 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py @@ -18,9 +18,11 @@ from __future__ import annotations import tempfile +import warnings from collections.abc import Sequence from typing import TYPE_CHECKING +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.google.cloud.hooks.gcs import GCSHook from airflow.providers.google.version_compat import BaseOperator @@ -59,6 +61,8 @@ class AzureBlobStorageToGCSOperator(BaseOperator): If set as a sequence, the identities from the list must grant Service Account Token Creator IAM role to the directly preceding identity, with first account from the list granting this role to the originating account. + :param return_gcs_uris: If True, return a list of GCS URIs. If False (default), return the legacy + string value and emit a deprecation warning. """ def __init__( @@ -73,6 +77,7 @@ def __init__( filename: str, gzip: bool, impersonation_chain: str | Sequence[str] | None = None, + return_gcs_uris: bool = False, **kwargs, ) -> None: super().__init__(**kwargs) @@ -85,6 +90,7 @@ def __init__( self.filename = filename self.gzip = gzip self.impersonation_chain = impersonation_chain + self.return_gcs_uris = return_gcs_uris template_fields: Sequence[str] = ( "blob_name", @@ -94,7 +100,7 @@ def __init__( "filename", ) - def execute(self, context: Context) -> list[str]: + def execute(self, context: Context) -> str | list[str]: azure_hook = WasbHook(wasb_conn_id=self.wasb_conn_id) gcs_hook = GCSHook( gcp_conn_id=self.gcp_conn_id, @@ -122,7 +128,16 @@ def execute(self, context: Context) -> list[str]: self.blob_name, self.bucket_name, ) - return [f"gs://{self.bucket_name}/{self.object_name}"] + if self.return_gcs_uris: + return [f"gs://{self.bucket_name}/{self.object_name}"] + else: + warnings.warn( + "Returning a string from AzureBlobStorageToGCSOperator is deprecated and will " + "change to list[str] in a future release. Set return_gcs_uris=True to opt in.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + return f"gs://{self.bucket_name}/{self.object_name}" def get_openlineage_facets_on_start(self): from airflow.providers.common.compat.openlineage.facet import Dataset diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py index eea4a8347f0cf..3dcffd29c8cb3 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py @@ -66,6 +66,8 @@ class AzureFileShareToGCSOperator(BaseOperator): If set as a sequence, the identities from the list must grant Service Account Token Creator IAM role to the directly preceding identity, with first account from the list granting this role to the originating account (templated). + :param return_gcs_uris: If True, return a list of GCS URIs. If False (default), return the legacy + list of Azure FileShare filenames and emit a deprecation warning. Note that ``share_name``, ``directory_path``, ``prefix``, and ``dest_gcs`` are templated, so you can use variables in them if you wish. @@ -92,6 +94,7 @@ def __init__( replace: bool = False, gzip: bool = False, google_impersonation_chain: str | Sequence[str] | None = None, + return_gcs_uris: bool = False, **kwargs, ): super().__init__(**kwargs) @@ -113,6 +116,7 @@ def __init__( self.replace = replace self.gzip = gzip self.google_impersonation_chain = google_impersonation_chain + self.return_gcs_uris = return_gcs_uris def _check_inputs(self) -> None: if self.dest_gcs and not gcs_object_is_directory(self.dest_gcs): @@ -188,4 +192,13 @@ def execute(self, context: Context) -> list[str]: self.log.info("There are no new files to sync. Have a nice day!") self.log.info("In sync, no files needed to be uploaded to Google Cloud Storage") - return uploaded_gcs_uris + if self.return_gcs_uris: + return uploaded_gcs_uris + else: + warnings.warn( + "Returning a list of Azure FileShare filenames from AzureFileShareToGCSOperator is deprecated and " + "will change to list[str] of GCS URIs in a future release. Set return_gcs_uris=True to opt in.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + return files diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py index d51d05885d90d..833857ef3c01f 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py @@ -18,8 +18,12 @@ from unittest import mock +import pytest + from airflow.providers.google.cloud.transfers.azure_blob_to_gcs import AzureBlobStorageToGCSOperator +pytestmark = pytest.mark.filterwarnings("ignore::airflow.exceptions.AirflowProviderDeprecationWarning") + WASB_CONN_ID = "wasb_default" GCP_CONN_ID = "google_cloud_default" BLOB_NAME = "azure_blob" @@ -57,10 +61,17 @@ def test_init(self): assert operator.impersonation_chain == IMPERSONATION_CHAIN assert operator.task_id == TASK_ID + @pytest.mark.parametrize( + "return_gcs_uris, expected", + [ + (False, f"gs://{BUCKET_NAME}/{OBJECT_NAME}"), + (True, [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"]), + ], + ) @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.WasbHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.GCSHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.tempfile") - def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb): + def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb, return_gcs_uris, expected): op = AzureBlobStorageToGCSOperator( wasb_conn_id=WASB_CONN_ID, gcp_conn_id=GCP_CONN_ID, @@ -71,6 +82,7 @@ def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb): filename=FILENAME, gzip=GZIP, impersonation_chain=IMPERSONATION_CHAIN, + return_gcs_uris=return_gcs_uris, task_id=TASK_ID, ) @@ -91,7 +103,7 @@ def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb): gzip=GZIP, filename=mock_temp.NamedTemporaryFile.return_value.__enter__.return_value.name, ) - assert result == [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"] + assert result == expected @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.WasbHook") def test_execute_single_file_transfer_openlineage(self, mock_hook_wasb): diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py index 31e9cfcc3cfc1..7cccf13fa252f 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py @@ -18,8 +18,12 @@ from unittest import mock +import pytest + from airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs import AzureFileShareToGCSOperator +pytestmark = pytest.mark.filterwarnings("ignore::airflow.exceptions.AirflowProviderDeprecationWarning") + TASK_ID = "test-azure-fileshare-to-gcs" AZURE_FILESHARE_SHARE = "test-share" AZURE_FILESHARE_DIRECTORY_PATH = "/path/to/dir" @@ -52,9 +56,10 @@ def test_init(self): assert operator.dest_gcs == GCS_PATH_PREFIX assert operator.google_impersonation_chain == IMPERSONATION_CHAIN + @pytest.mark.parametrize("return_gcs_uris", [True, False]) @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.GCSHook") - def test_execute(self, gcs_mock_hook, azure_fileshare_mock_hook): + def test_execute(self, gcs_mock_hook, azure_fileshare_mock_hook, return_gcs_uris): """Test the execute function when the run is successful.""" operator = AzureFileShareToGCSOperator( @@ -65,6 +70,7 @@ def test_execute(self, gcs_mock_hook, azure_fileshare_mock_hook): gcp_conn_id=GCS_CONN_ID, dest_gcs=GCS_PATH_PREFIX, google_impersonation_chain=IMPERSONATION_CHAIN, + return_gcs_uris=return_gcs_uris, ) azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES @@ -87,7 +93,11 @@ def test_execute(self, gcs_mock_hook, azure_fileshare_mock_hook): impersonation_chain=IMPERSONATION_CHAIN, ) - expected_files = [f"gs://gcs-bucket/data/{file_name}" for file_name in MOCK_FILES] + expected_files = ( + [f"gs://gcs-bucket/data/{file_name}" for file_name in MOCK_FILES] + if return_gcs_uris + else MOCK_FILES + ) assert sorted(expected_files) == sorted(uploaded_files) @mock.patch("airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs.AzureFileShareHook") @@ -104,6 +114,7 @@ def test_execute_with_gzip(self, gcs_mock_hook, azure_fileshare_mock_hook): dest_gcs=GCS_PATH_PREFIX, google_impersonation_chain=IMPERSONATION_CHAIN, gzip=True, + return_gcs_uris=True, ) azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES From 17d819219e3d196b133dfe0ad93ef1e6bd142620 Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 25 Jan 2026 15:41:49 -0800 Subject: [PATCH 3/7] remove `return_gcs_uris` parameter from test_execute_with_gzip --- .../unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py | 1 - 1 file changed, 1 deletion(-) diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py index 7cccf13fa252f..261b58af968c2 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py @@ -114,7 +114,6 @@ def test_execute_with_gzip(self, gcs_mock_hook, azure_fileshare_mock_hook): dest_gcs=GCS_PATH_PREFIX, google_impersonation_chain=IMPERSONATION_CHAIN, gzip=True, - return_gcs_uris=True, ) azure_fileshare_mock_hook.return_value.list_files.return_value = MOCK_FILES From 81d9b13e616897f3ab035925507abc2f604f08f9 Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 25 Jan 2026 20:50:22 -0800 Subject: [PATCH 4/7] Change style to early return to fix CI error --- .../google/cloud/transfers/azure_blob_to_gcs.py | 15 +++++++-------- .../cloud/transfers/azure_fileshare_to_gcs.py | 15 +++++++-------- 2 files changed, 14 insertions(+), 16 deletions(-) diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py index a34ceed789f3f..89c6e00ffe4e9 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py @@ -130,14 +130,13 @@ def execute(self, context: Context) -> str | list[str]: ) if self.return_gcs_uris: return [f"gs://{self.bucket_name}/{self.object_name}"] - else: - warnings.warn( - "Returning a string from AzureBlobStorageToGCSOperator is deprecated and will " - "change to list[str] in a future release. Set return_gcs_uris=True to opt in.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) - return f"gs://{self.bucket_name}/{self.object_name}" + warnings.warn( + "Returning a string from AzureBlobStorageToGCSOperator is deprecated and will " + "change to list[str] in a future release. Set return_gcs_uris=True to opt in.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + return f"gs://{self.bucket_name}/{self.object_name}" def get_openlineage_facets_on_start(self): from airflow.providers.common.compat.openlineage.facet import Dataset diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py index 3dcffd29c8cb3..3a6aeb262f11b 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py @@ -194,11 +194,10 @@ def execute(self, context: Context) -> list[str]: if self.return_gcs_uris: return uploaded_gcs_uris - else: - warnings.warn( - "Returning a list of Azure FileShare filenames from AzureFileShareToGCSOperator is deprecated and " - "will change to list[str] of GCS URIs in a future release. Set return_gcs_uris=True to opt in.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) - return files + warnings.warn( + "Returning a list of Azure FileShare filenames from AzureFileShareToGCSOperator is deprecated and " + "will change to list[str] of GCS URIs in a future release. Set return_gcs_uris=True to opt in.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + return files From cf297b21127760989981214bdb11cfaa7d2746e3 Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 25 Jan 2026 22:03:08 -0800 Subject: [PATCH 5/7] Fix parameterization syntax in test for AzureBlobStorageToGCSOperator --- .../tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py index 833857ef3c01f..6b53896a879d6 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py @@ -62,7 +62,7 @@ def test_init(self): assert operator.task_id == TASK_ID @pytest.mark.parametrize( - "return_gcs_uris, expected", + ("return_gcs_uris", "expected"), [ (False, f"gs://{BUCKET_NAME}/{OBJECT_NAME}"), (True, [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"]), From 6fa964c415ddcf5775efc27f7139d49551559470 Mon Sep 17 00:00:00 2001 From: nailo2c Date: Sun, 1 Feb 2026 10:46:33 -0800 Subject: [PATCH 6/7] Refactor AzureBlobStorageToGCSOperator and AzureFileShareToGCSOperator to use unwrap_single parameter for GCS URI returns; update tests accordingly --- .../cloud/transfers/azure_blob_to_gcs.py | 36 +++++++++++-------- .../cloud/transfers/azure_fileshare_to_gcs.py | 25 +++++++------ .../cloud/transfers/test_azure_blob_to_gcs.py | 12 +++---- .../transfers/test_azure_fileshare_to_gcs.py | 2 +- 4 files changed, 42 insertions(+), 33 deletions(-) diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py index 89c6e00ffe4e9..536fa85375edc 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_blob_to_gcs.py @@ -18,11 +18,9 @@ from __future__ import annotations import tempfile -import warnings from collections.abc import Sequence from typing import TYPE_CHECKING -from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.google.cloud.hooks.gcs import GCSHook from airflow.providers.google.version_compat import BaseOperator @@ -61,8 +59,10 @@ class AzureBlobStorageToGCSOperator(BaseOperator): If set as a sequence, the identities from the list must grant Service Account Token Creator IAM role to the directly preceding identity, with first account from the list granting this role to the originating account. - :param return_gcs_uris: If True, return a list of GCS URIs. If False (default), return the legacy - string value and emit a deprecation warning. + :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. """ def __init__( @@ -77,7 +77,7 @@ def __init__( filename: str, gzip: bool, impersonation_chain: str | Sequence[str] | None = None, - return_gcs_uris: bool = False, + unwrap_single: bool | None = None, **kwargs, ) -> None: super().__init__(**kwargs) @@ -90,7 +90,18 @@ def __init__( self.filename = filename self.gzip = gzip self.impersonation_chain = impersonation_chain - self.return_gcs_uris = return_gcs_uris + if unwrap_single is None: + self.unwrap_single = True + import warnings + + warnings.warn( + "The default value of `unwrap_single` will change from True to False in a future release. " + "Please set `unwrap_single` explicitly to avoid this warning.", + FutureWarning, + stacklevel=2, + ) + else: + self.unwrap_single = unwrap_single template_fields: Sequence[str] = ( "blob_name", @@ -128,15 +139,10 @@ def execute(self, context: Context) -> str | list[str]: self.blob_name, self.bucket_name, ) - if self.return_gcs_uris: - return [f"gs://{self.bucket_name}/{self.object_name}"] - warnings.warn( - "Returning a string from AzureBlobStorageToGCSOperator is deprecated and will " - "change to list[str] in a future release. Set return_gcs_uris=True to opt in.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) - return f"gs://{self.bucket_name}/{self.object_name}" + gcs_uri = f"gs://{self.bucket_name}/{self.object_name}" + if self.unwrap_single: + return gcs_uri + return [gcs_uri] def get_openlineage_facets_on_start(self): from airflow.providers.common.compat.openlineage.facet import Dataset diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py index 3a6aeb262f11b..993ce091d5010 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py @@ -94,7 +94,7 @@ def __init__( replace: bool = False, gzip: bool = False, google_impersonation_chain: str | Sequence[str] | None = None, - return_gcs_uris: bool = False, + return_gcs_uris: bool | None = None, **kwargs, ): super().__init__(**kwargs) @@ -116,7 +116,16 @@ def __init__( self.replace = replace self.gzip = gzip self.google_impersonation_chain = google_impersonation_chain - self.return_gcs_uris = return_gcs_uris + if return_gcs_uris is None: + self.return_gcs_uris = False + warnings.warn( + "Returning a list of Azure FileShare filenames from AzureFileShareToGCSOperator 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, + ) + else: + self.return_gcs_uris = return_gcs_uris def _check_inputs(self) -> None: if self.dest_gcs and not gcs_object_is_directory(self.dest_gcs): @@ -192,12 +201,6 @@ def execute(self, context: Context) -> list[str]: self.log.info("There are no new files to sync. Have a nice day!") self.log.info("In sync, no files needed to be uploaded to Google Cloud Storage") - if self.return_gcs_uris: - return uploaded_gcs_uris - warnings.warn( - "Returning a list of Azure FileShare filenames from AzureFileShareToGCSOperator is deprecated and " - "will change to list[str] of GCS URIs in a future release. Set return_gcs_uris=True to opt in.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) - return files + if not self.return_gcs_uris: + return files + return uploaded_gcs_uris diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py index 6b53896a879d6..ae3300af988b9 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_blob_to_gcs.py @@ -22,7 +22,7 @@ from airflow.providers.google.cloud.transfers.azure_blob_to_gcs import AzureBlobStorageToGCSOperator -pytestmark = pytest.mark.filterwarnings("ignore::airflow.exceptions.AirflowProviderDeprecationWarning") +pytestmark = pytest.mark.filterwarnings("ignore::FutureWarning") WASB_CONN_ID = "wasb_default" GCP_CONN_ID = "google_cloud_default" @@ -62,16 +62,16 @@ def test_init(self): assert operator.task_id == TASK_ID @pytest.mark.parametrize( - ("return_gcs_uris", "expected"), + ("unwrap_single", "expected"), [ - (False, f"gs://{BUCKET_NAME}/{OBJECT_NAME}"), - (True, [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"]), + (True, f"gs://{BUCKET_NAME}/{OBJECT_NAME}"), + (False, [f"gs://{BUCKET_NAME}/{OBJECT_NAME}"]), ], ) @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.WasbHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.GCSHook") @mock.patch("airflow.providers.google.cloud.transfers.azure_blob_to_gcs.tempfile") - def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb, return_gcs_uris, expected): + def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb, unwrap_single, expected): op = AzureBlobStorageToGCSOperator( wasb_conn_id=WASB_CONN_ID, gcp_conn_id=GCP_CONN_ID, @@ -82,7 +82,7 @@ def test_execute(self, mock_temp, mock_hook_gcs, mock_hook_wasb, return_gcs_uris filename=FILENAME, gzip=GZIP, impersonation_chain=IMPERSONATION_CHAIN, - return_gcs_uris=return_gcs_uris, + unwrap_single=unwrap_single, task_id=TASK_ID, ) diff --git a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py index 261b58af968c2..a70d7cc709b7d 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_azure_fileshare_to_gcs.py @@ -22,7 +22,7 @@ from airflow.providers.google.cloud.transfers.azure_fileshare_to_gcs import AzureFileShareToGCSOperator -pytestmark = pytest.mark.filterwarnings("ignore::airflow.exceptions.AirflowProviderDeprecationWarning") +pytestmark = pytest.mark.filterwarnings("ignore::FutureWarning") TASK_ID = "test-azure-fileshare-to-gcs" AZURE_FILESHARE_SHARE = "test-share" From ea4b39d6107bf39656d8e13e10b5c7b1f6498d8e Mon Sep 17 00:00:00 2001 From: nailo2c Date: Fri, 6 Feb 2026 16:17:46 -0800 Subject: [PATCH 7/7] Fix dest_gcs in example_azure_fileshare_to_gcs.py: add gs:// prefix and trailing / required by _parse_gcs_url --- .../system/google/cloud/azure/example_azure_fileshare_to_gcs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/providers/google/tests/system/google/cloud/azure/example_azure_fileshare_to_gcs.py b/providers/google/tests/system/google/cloud/azure/example_azure_fileshare_to_gcs.py index d03c493c8dd9c..c62db075d66ab 100644 --- a/providers/google/tests/system/google/cloud/azure/example_azure_fileshare_to_gcs.py +++ b/providers/google/tests/system/google/cloud/azure/example_azure_fileshare_to_gcs.py @@ -62,7 +62,7 @@ sync_azure_files_with_gcs = AzureFileShareToGCSOperator( task_id="sync_azure_files_with_gcs", share_name=AZURE_SHARE_NAME, - dest_gcs=BUCKET_NAME, + dest_gcs=f"gs://{BUCKET_NAME}/", directory_path=AZURE_DIRECTORY_PATH, replace=False, gzip=True,