diff --git a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py index aa027e71d854a..b6f7e4f372908 100644 --- a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py @@ -217,28 +217,10 @@ def __init__( super().__init__(**kwargs) self.source_bucket = source_bucket - if source_object and WILDCARD in source_object: - warnings.warn( - "Usage of wildcard (*) in 'source_object' is deprecated, utilize 'match_glob' instead. Planned removal date: October 5, 2026.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) self.source_object = source_object - if source_objects and any(WILDCARD in obj for obj in source_objects): - warnings.warn( - "Usage of wildcard (*) in 'source_objects' is deprecated, utilize 'match_glob' instead. Planned removal date: October 5, 2026.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) self.source_objects = source_objects self.destination_bucket = destination_bucket self.destination_object = destination_object - if delimiter: - warnings.warn( - "Usage of 'delimiter' is deprecated, please use 'match_glob' instead. Planned removal date: October 5, 2026.", - AirflowProviderDeprecationWarning, - stacklevel=2, - ) self.delimiter = delimiter self.move_object = move_object self.replace = replace @@ -253,7 +235,29 @@ def __init__( self.retain_until_time = retain_until_time self.retention_mode = retention_mode + def _warn_on_deprecated_template_fields(self) -> None: + if self.source_object and WILDCARD in self.source_object: + warnings.warn( + "Usage of wildcard (*) in 'source_object' is deprecated, utilize 'match_glob' instead. Planned removal date: October 5, 2026.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + if self.source_objects and any(WILDCARD in obj for obj in self.source_objects): + warnings.warn( + "Usage of wildcard (*) in 'source_objects' is deprecated, utilize 'match_glob' instead. Planned removal date: October 5, 2026.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + if self.delimiter: + warnings.warn( + "Usage of 'delimiter' is deprecated, please use 'match_glob' instead. Planned removal date: October 5, 2026.", + AirflowProviderDeprecationWarning, + stacklevel=2, + ) + def execute(self, context: Context) -> list[str]: + self._warn_on_deprecated_template_fields() + hook = GCSHook( gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain, diff --git a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py index 7bbfa1c2285b4..bfd2c8f586fa7 100644 --- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py +++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py @@ -24,7 +24,7 @@ from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.openlineage.facet import Dataset -from airflow.providers.common.compat.sdk import AirflowException +from airflow.providers.common.compat.sdk import DAG, AirflowException from airflow.providers.google.cloud.transfers.gcs_to_gcs import WILDCARD, GCSToGCSOperator TASK_ID = "test-gcs-to-gcs-operator" @@ -1035,25 +1035,19 @@ def test_copy_files_into_a_folder( def test_get_openlineage_facets_on_complete( self, mock_hook, source_objects, destination_object, inputs, outputs ): + operator = GCSToGCSOperator( + task_id=TASK_ID, + source_bucket=TEST_BUCKET, + source_objects=source_objects, + destination_bucket=DESTINATION_BUCKET, + destination_object=destination_object, + ) + if source_objects and any(WILDCARD in obj for obj in source_objects): with pytest.warns(AirflowProviderDeprecationWarning, match="Usage of wildcard"): - operator = GCSToGCSOperator( - task_id=TASK_ID, - source_bucket=TEST_BUCKET, - source_objects=source_objects, - destination_bucket=DESTINATION_BUCKET, - destination_object=destination_object, - ) + operator.execute(None) else: - operator = GCSToGCSOperator( - task_id=TASK_ID, - source_bucket=TEST_BUCKET, - source_objects=source_objects, - destination_bucket=DESTINATION_BUCKET, - destination_object=destination_object, - ) - - operator.execute(None) + operator.execute(None) lineage = operator.get_openlineage_facets_on_complete(None) assert len(lineage.inputs) == len(inputs) @@ -1082,15 +1076,15 @@ def test_execute_returns_list_of_destination_uris_single_file(self, mock_hook): @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook") def test_execute_returns_list_of_destination_uris_multiple_files(self, mock_hook): mock_hook.return_value.list.return_value = SOURCE_OBJECTS_LIST + operator = GCSToGCSOperator( + task_id=TASK_ID, + source_bucket=TEST_BUCKET, + source_object=SOURCE_OBJECT_WILDCARD_FILENAME, + destination_bucket=DESTINATION_BUCKET, + destination_object=DESTINATION_OBJECT_PREFIX, + ) with pytest.warns(AirflowProviderDeprecationWarning, match="Usage of wildcard"): - operator = GCSToGCSOperator( - task_id=TASK_ID, - source_bucket=TEST_BUCKET, - source_object=SOURCE_OBJECT_WILDCARD_FILENAME, - destination_bucket=DESTINATION_BUCKET, - destination_object=DESTINATION_OBJECT_PREFIX, - ) - result = operator.execute(None) + result = operator.execute(None) expected = [ f"gs://{DESTINATION_BUCKET}/foo/bar/file1.txt", f"gs://{DESTINATION_BUCKET}/foo/bar/file2.txt", @@ -1098,6 +1092,25 @@ def test_execute_returns_list_of_destination_uris_multiple_files(self, mock_hook ] assert sorted(result) == sorted(expected) + @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook") + def test_wildcard_deprecation_warning_uses_rendered_source_object(self, mock_hook): + mock_hook.return_value.list.return_value = SOURCE_OBJECTS_LIST + with DAG(dag_id="test_gcs_to_gcs_wildcard_templating", start_date=datetime(2024, 1, 1)) as dag: + operator = GCSToGCSOperator( + task_id=TASK_ID, + source_bucket=TEST_BUCKET, + source_object="{{ params.source_object }}", + destination_bucket=DESTINATION_BUCKET, + destination_object=DESTINATION_OBJECT_PREFIX, + dag=dag, + ) + + operator.render_template_fields({"params": {"source_object": SOURCE_OBJECT_WILDCARD_FILENAME}}) + assert operator.source_object == SOURCE_OBJECT_WILDCARD_FILENAME + + with pytest.warns(AirflowProviderDeprecationWarning, match="Usage of wildcard"): + operator.execute(mock.MagicMock()) + @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook") def test_execute_returns_empty_list_when_no_files_copied(self, mock_hook): mock_hook.return_value.is_updated_after.return_value = False diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index 76ba180595a56..8193b425d561f 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -26,7 +26,6 @@ providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::C providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py::AzureFileShareToGCSOperator providers/google/src/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py::BigQueryToMsSqlOperator providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py::GCSToBigQueryOperator -providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py::GCSToGCSOperator providers/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py::GoogleCampaignManagerDeleteReportOperator providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator