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 f2802c459fe Preserve Dataform workflow invocation config (#53843)
(#69161)
f2802c459fe is described below
commit f2802c459fe4c2012ba058035b812296b49fa0ed
Author: Deepak Jain <[email protected]>
AuthorDate: Wed Jul 22 22:53:25 2026 +0530
Preserve Dataform workflow invocation config (#53843) (#69161)
---
.../providers/google/cloud/hooks/dataform.py | 2 ++
.../tests/unit/google/cloud/hooks/test_dataform.py | 23 +++++++++++++++++++++-
2 files changed, 24 insertions(+), 1 deletion(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
b/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
index 48162f1fd8c..9167689daa8 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
@@ -187,6 +187,8 @@ class DataformHook(GoogleBaseHook):
"""
client = self.get_dataform_client()
parent =
f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
+ if isinstance(workflow_invocation, dict):
+ workflow_invocation = WorkflowInvocation(workflow_invocation)
return client.create_workflow_invocation(
request={"parent": parent, "workflow_invocation":
workflow_invocation},
retry=retry,
diff --git a/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
b/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
index 2dda5ffe431..c70492a7784 100644
--- a/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
+++ b/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
@@ -50,6 +50,12 @@ WORKFLOW_INVOCATION = {
f"{REPOSITORY_ID}/compilationResults/{COMPILATION_RESULT_ID}"
),
}
+FULL_REFRESH_WORKFLOW_INVOCATION = {
+ **WORKFLOW_INVOCATION,
+ "invocation_config": {
+ "fully_refresh_incremental_tables_enabled": True,
+ },
+}
WORKFLOW_INVOCATION_ID = "test_workflow_invocation_id"
PATH_TO_FOLDER = "path/to/folder"
FILEPATH = "path/to/file.txt"
@@ -113,12 +119,27 @@ class TestDataformHook:
)
parent =
f"projects/{PROJECT_ID}/locations/{REGION}/repositories/{REPOSITORY_ID}"
mock_client.return_value.create_workflow_invocation.assert_called_once_with(
- request=dict(parent=parent,
workflow_invocation=WORKFLOW_INVOCATION),
+ request=dict(parent=parent,
workflow_invocation=WorkflowInvocation(WORKFLOW_INVOCATION)),
retry=DEFAULT,
timeout=None,
metadata=(),
)
+ @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+ def
test_create_workflow_invocation_preserves_invocation_config_from_dict(self,
mock_client):
+ self.hook.create_workflow_invocation(
+ project_id=PROJECT_ID,
+ region=REGION,
+ repository_id=REPOSITORY_ID,
+ workflow_invocation=FULL_REFRESH_WORKFLOW_INVOCATION,
+ )
+
+ request =
mock_client.return_value.create_workflow_invocation.call_args.kwargs["request"]
+ workflow_invocation = request["workflow_invocation"]
+ assert isinstance(workflow_invocation, WorkflowInvocation)
+ assert workflow_invocation.compilation_result ==
WORKFLOW_INVOCATION["compilation_result"]
+ assert
workflow_invocation.invocation_config.fully_refresh_incremental_tables_enabled
is True
+
@mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
def test_get_workflow_invocation(self, mock_client):
self.hook.get_workflow_invocation(
