You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2026/09/20 17:58:22 UTC
(airflow) branch main updated: Resolve GCSFileTransformOperator destination fallbacks after rendering (#70488)
This is an automated email from the ASF dual-hosted git repository.
potiuk 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 e75094dbafc Resolve GCSFileTransformOperator destination fallbacks after rendering (#70488)
e75094dbafc is described below
commit e75094dbafcaa653fc7934a99b2a9be4f027b7ef
Author: Dr Alex Mitre <be...@hotmail.com>
AuthorDate: Sun Sep 20 11:58:10 2026 -0600
Resolve GCSFileTransformOperator destination fallbacks after rendering (#70488)
* Resolve GCSFileTransformOperator destination fallbacks after rendering
destination_bucket/destination_object are template fields, so resolving
their 'or source_*' fallbacks in __init__ reads the unrendered value.
Move the fallback resolution to execute() where the rendered values are
available, keep get_openlineage_facets_on_complete resolving fallbacks
locally (it can run outside execute), and remove the operator from the
validate_operators_init exemptions list.
Related: #70296
* Add tests for the GCSFileTransformOperator destination fallback
Cover the omitted-destination path in execute() and
get_openlineage_facets_on_start(): the source values are changed after
construction (as template rendering does) and the destination must follow
them. Both tests fail against the previous __init__-time fallback.
Generated-by: Claude Opus 5
---------
Co-authored-by: Dr Alex Mitre <30...@users.noreply.github.com>
Co-authored-by: Jarek Potiuk <ja...@potiuk.com>
---
.../providers/google/cloud/operators/gcs.py | 10 ++--
.../tests/unit/google/cloud/operators/test_gcs.py | 55 ++++++++++++++++++++++
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 61 insertions(+), 5 deletions(-)
diff --git a/providers/google/src/airflow/providers/google/cloud/operators/gcs.py b/providers/google/src/airflow/providers/google/cloud/operators/gcs.py
index 2cc26532148..59e077e2406 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/gcs.py
+++ b/providers/google/src/airflow/providers/google/cloud/operators/gcs.py
@@ -692,8 +692,8 @@ class GCSFileTransformOperator(GoogleCloudBaseOperator):
super().__init__(**kwargs)
self.source_bucket = source_bucket
self.source_object = source_object
- self.destination_bucket = destination_bucket or self.source_bucket
- self.destination_object = destination_object or self.source_object
+ self.destination_bucket = destination_bucket
+ self.destination_object = destination_object
self.gcp_conn_id = gcp_conn_id
self.transform_script = transform_script
@@ -701,6 +701,8 @@ class GCSFileTransformOperator(GoogleCloudBaseOperator):
self.impersonation_chain = impersonation_chain
def execute(self, context: Context) -> None:
+ self.destination_bucket = self.destination_bucket or self.source_bucket
+ self.destination_object = self.destination_object or self.source_object
hook = GCSHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
with NamedTemporaryFile() as source_file, NamedTemporaryFile() as destination_file:
@@ -747,8 +749,8 @@ class GCSFileTransformOperator(GoogleCloudBaseOperator):
name=self.source_object,
)
output_dataset = Dataset(
- namespace=f"gs://{self.destination_bucket}",
- name=self.destination_object,
+ namespace=f"gs://{self.destination_bucket or self.source_bucket}",
+ name=self.destination_object or self.source_object,
)
return OperatorLineage(inputs=[input_dataset], outputs=[output_dataset])
diff --git a/providers/google/tests/unit/google/cloud/operators/test_gcs.py b/providers/google/tests/unit/google/cloud/operators/test_gcs.py
index ba1c8ef9e79..4d53d254e49 100644
--- a/providers/google/tests/unit/google/cloud/operators/test_gcs.py
+++ b/providers/google/tests/unit/google/cloud/operators/test_gcs.py
@@ -426,6 +426,61 @@ class TestGCSFileTransformOperator:
assert lineage.inputs[0] == expected_input
assert lineage.outputs[0] == expected_output
+ @mock.patch("airflow.providers.google.cloud.operators.gcs.NamedTemporaryFile")
+ @mock.patch("airflow.providers.google.cloud.operators.gcs.subprocess")
+ @mock.patch("airflow.providers.google.cloud.operators.gcs.GCSHook")
+ def test_execute_destination_falls_back_to_rendered_source(
+ self, mock_hook, mock_subprocess, mock_tempfile
+ ):
+ source = "source"
+ destination = "destination"
+ mock1 = mock.Mock()
+ mock2 = mock.Mock()
+ mock1.name = source
+ mock2.name = destination
+ mock_tempfile.return_value.__enter__.side_effect = [mock1, mock2]
+
+ mock_proc = mock.MagicMock()
+ mock_proc.returncode = 0
+ mock_proc.stdout.readline = lambda: b""
+ mock_proc.wait.return_value = None
+ mock_subprocess.Popen.return_value.__enter__.return_value = mock_proc
+
+ op = GCSFileTransformOperator(
+ task_id=TASK_ID,
+ source_bucket="{{ var.value.bucket }}",
+ source_object="{{ var.value.object }}",
+ transform_script="script.py",
+ )
+ # Template rendering happens after __init__; the destination fallback must see the rendered values.
+ op.source_bucket = TEST_BUCKET
+ op.source_object = "rendered.txt"
+
+ op.execute(context=mock.MagicMock())
+
+ mock_hook.return_value.download.assert_called_once_with(
+ bucket_name=TEST_BUCKET, object_name="rendered.txt", filename=source
+ )
+ mock_hook.return_value.upload.assert_called_with(
+ bucket_name=TEST_BUCKET, object_name="rendered.txt", filename=destination
+ )
+
+ def test_get_openlineage_facets_on_start_destination_falls_back_to_rendered_source(self):
+ operator = GCSFileTransformOperator(
+ task_id=TASK_ID,
+ source_bucket="{{ var.value.bucket }}",
+ source_object="{{ var.value.object }}",
+ transform_script="/path/to_script",
+ )
+ operator.source_bucket = TEST_BUCKET
+ operator.source_object = "folder/a.txt"
+
+ lineage = operator.get_openlineage_facets_on_start()
+
+ expected = Dataset(namespace=f"gs://{TEST_BUCKET}", name="folder/a.txt")
+ assert lineage.inputs == [expected]
+ assert lineage.outputs == [expected]
+
class TestGCSTimeSpanFileTransformOperatorDateInterpolation:
def test_execute(self):
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 952e1576cc1..117ee0886e5 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -12,6 +12,5 @@ providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_tran
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocSubmitJobOperator
providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator
-providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTransformOperator
providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor