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 7c47cc44383 Validate GCP storage transfer job body after template
rendering (#70529)
7c47cc44383 is described below
commit 7c47cc443836f02944703a3d29514eabc58b0f51
Author: Dr Alex Mitre <[email protected]>
AuthorDate: Wed Sep 23 01:35:20 2026 -0600
Validate GCP storage transfer job body after template rendering (#70529)
Co-authored-by: Shahar Epstein <[email protected]>
---
.../operators/cloud_storage_transfer_service.py | 18 +++++++--------
.../test_cloud_storage_transfer_service.py | 27 +++++++++++++++++++++-
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 35 insertions(+), 11 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
b/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
index 85450a08d0b..7e74db4ece2 100644
---
a/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
+++
b/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
@@ -211,7 +211,7 @@ class
CloudDataTransferServiceCreateJobOperator(GoogleCloudBaseOperator):
For more information on how to use this operator, take a look at the
guide:
:ref:`howto/operator:CloudDataTransferServiceCreateJobOperator`
- :param body: (Required) The request body, as described in
+ :param body: (Required) The request body (templated), as described in
https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs#TransferJob
With three additional improvements:
@@ -220,6 +220,9 @@ class
CloudDataTransferServiceCreateJobOperator(GoogleCloudBaseOperator):
* credentials to Amazon Web Service should be stored in the connection
and indicated by the
aws_conn_id parameter
+ If the whole body is passed as a Jinja expression, set
``render_template_as_native_obj=True``
+ on the Dag; a default Jinja render would produce a string instead of a
mapping.
+
:param aws_conn_id: The connection ID used to retrieve credentials to
Amazon Web Service.
:param gcp_conn_id: The connection ID used to connect to Google Cloud.
@@ -257,26 +260,23 @@ class
CloudDataTransferServiceCreateJobOperator(GoogleCloudBaseOperator):
) -> None:
super().__init__(**kwargs)
self.body = body
- if isinstance(self.body, dict):
- self.body = deepcopy(body)
self.aws_conn_id = aws_conn_id
self.gcp_conn_id = gcp_conn_id
self.api_version = api_version
self.project_id = project_id
self.google_impersonation_chain = google_impersonation_chain
- self._validate_inputs()
-
- def _validate_inputs(self) -> None:
- TransferJobValidator(body=self.body).validate_body()
def execute(self, context: Context) -> dict:
- TransferJobPreprocessor(body=self.body,
aws_conn_id=self.aws_conn_id).process_body()
+ # TransferJobPreprocessor mutates the body in place, so copy it to
avoid mutating the caller's dict.
+ body = deepcopy(self.body) if isinstance(self.body, dict) else
self.body
+ TransferJobValidator(body=body).validate_body()
+ TransferJobPreprocessor(body=body,
aws_conn_id=self.aws_conn_id).process_body()
hook = CloudDataTransferServiceHook(
api_version=self.api_version,
gcp_conn_id=self.gcp_conn_id,
impersonation_chain=self.google_impersonation_chain,
)
- result = hook.create_transfer_job(body=self.body)
+ result = hook.create_transfer_job(body=body)
project_id = self.project_id or hook.project_id
if project_id:
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
b/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
index 3b24428ac1b..2a78b40d7fe 100644
---
a/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
+++
b/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
@@ -25,7 +25,7 @@ import pytest
import time_machine
from botocore.credentials import Credentials
-from airflow.providers.common.compat.sdk import AirflowException,
TaskDeferred, timezone
+from airflow.providers.common.compat.sdk import DAG, AirflowException,
TaskDeferred, timezone
from airflow.providers.google.cloud.hooks.cloud_storage_transfer_service
import (
ACCESS_KEY_ID,
AWS_ACCESS_KEY,
@@ -282,6 +282,29 @@ class TestTransferJobValidator:
class TestGcpStorageTransferJobCreateOperator:
+ @mock.patch(
+
"airflow.providers.google.cloud.operators.cloud_storage_transfer_service.CloudDataTransferServiceHook"
+ )
+
@mock.patch("airflow.providers.google.cloud.operators.cloud_storage_transfer_service.AwsBaseHook")
+ def test_templated_body_validated_at_execute_time(self, aws_hook,
mock_hook):
+ with DAG(
+ dag_id="test_transfer_native_render",
+ start_date=DEFAULT_DATE,
+ schedule=None,
+ render_template_as_native_obj=True,
+ ) as dag:
+ op = CloudDataTransferServiceCreateJobOperator(body="{{ body }}",
task_id=TASK_ID, dag=dag)
+
+ # Real Jinja rendering turns the "{{ body }}" expression into the dict
below,
+ # proving validation runs against the rendered value, not the template.
+ op.render_template_fields(
+ context={"body": {TRANSFER_SPEC: {AWS_S3_DATA_SOURCE:
{AWS_ACCESS_KEY: TEST_AWS_ACCESS_KEY}}}}
+ )
+
+ with pytest.raises(AirflowException, match="AWS credentials detected
inside the body parameter"):
+ op.execute(context=mock.MagicMock())
+ mock_hook.return_value.create_transfer_job.assert_not_called()
+
@mock.patch(
"airflow.providers.google.cloud.operators.cloud_storage_transfer_service.CloudDataTransferServiceHook"
)
@@ -317,6 +340,7 @@ class TestGcpStorageTransferJobCreateOperator:
)
body = deepcopy(VALID_TRANSFER_JOB_AWS)
del body["name"]
+ original_body = deepcopy(body)
op = CloudDataTransferServiceCreateJobOperator(
body=body,
task_id=TASK_ID,
@@ -334,6 +358,7 @@ class TestGcpStorageTransferJobCreateOperator:
mock_hook.return_value.create_transfer_job.assert_called_once_with(body=VALID_TRANSFER_JOB_AWS_RAW)
assert result == VALID_TRANSFER_JOB_AWS
+ assert body == original_body
@mock.patch(
"airflow.providers.google.cloud.operators.cloud_storage_transfer_service.CloudDataTransferServiceHook"
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index afdc7deba3b..ada2c476766 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -8,5 +8,4 @@
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStartDbClusterOperator
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStopDbClusterOperator
providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
-providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py::CloudDataTransferServiceCreateJobOperator
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
