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 21ed9fc97be Replace vars(response) with attribute access in Azure Data
Factory and Synapse operators for SDK v10 compatibility (#69689)
21ed9fc97be is described below
commit 21ed9fc97be0f18ed243c5a61d48e1ba0b11a7d7
Author: Malte Niederstadt <[email protected]>
AuthorDate: Sun Jul 19 23:53:33 2026 +0200
Replace vars(response) with attribute access in Azure Data Factory and
Synapse operators for SDK v10 compatibility (#69689)
* fix(providers/azure): use attribute access instead of vars() for SDK v10
hybrid model compat
* added regression tests for azure data factory and azure synapse operators
* Fix ruff formatting in test_synapse.py
---------
Co-authored-by: Malte Niederstadt <[email protected]>
---
.../microsoft/azure/operators/data_factory.py | 2 +-
.../providers/microsoft/azure/operators/synapse.py | 4 +-
.../microsoft/azure/operators/test_data_factory.py | 26 ++++++++++++
.../unit/microsoft/azure/operators/test_synapse.py | 46 ++++++++++++++++++++++
4 files changed, 75 insertions(+), 3 deletions(-)
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
index 495eb27d6a7..36321dbe10a 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
@@ -181,7 +181,7 @@ class AzureDataFactoryRunPipelineOperator(BaseOperator):
start_from_failure=self.start_from_failure,
parameters=self.parameters,
)
- self.run_id = vars(response)["run_id"]
+ self.run_id = response.run_id
# Push the ``run_id`` value to XCom regardless of what happens during
execution. This allows for
# retrieval the executed pipeline's ``run_id`` for downstream tasks
especially if performing an
# asynchronous wait.
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
index b0c171a13bd..19a082ced77 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
@@ -104,7 +104,7 @@ class AzureSynapseRunSparkBatchOperator(BaseOperator):
self.log.info("Executing the Synapse spark job.")
response = self.hook.run_spark_job(payload=self.payload)
self.log.info(response)
- self.job_id = vars(response)["id"]
+ self.job_id = response.id
# Push the ``job_id`` value to XCom regardless of what happens during
execution. This allows for
# retrieval the executed job's ``id`` for downstream tasks especially
if performing an
# asynchronous wait.
@@ -257,7 +257,7 @@ class AzureSynapseRunPipelineOperator(BaseOperator):
start_activity_name=self.start_activity_name,
parameters=self.parameters,
)
- self.run_id = vars(response)["run_id"]
+ self.run_id = response.run_id
# Push the ``run_id`` value to XCom regardless of what happens during
execution. This allows for
# retrieval the executed pipeline's ``run_id`` for downstream tasks
especially if performing an
# asynchronous wait.
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
index c6403030480..7804891fd4b 100644
---
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
@@ -235,6 +235,32 @@ class TestAzureDataFactoryRunPipelineOperator:
# Checking the pipeline run status should _not_ be called when
``wait_for_termination`` is False.
mock_get_pipeline_run.assert_not_called()
+
@mock.patch("airflow.providers.microsoft.azure.hooks.data_factory.AzureDataFactoryHook.run_pipeline")
+ def test_run_id_extracted_from_hybrid_model_response(self,
mock_run_pipeline):
+ """Regression test: azure-mgmt-datafactory v10 hybrid models don't
expose attributes via vars().
+
+ Hybrid models use property descriptors, so vars(response)["run_id"]
raises KeyError.
+ The operator must use attribute access (response.run_id) which works
with both old and new models.
+ """
+
+ class HybridModelResponse(dict):
+ """Simulates an azure-mgmt-datafactory v10 hybrid model
response."""
+
+ def __init__(self):
+ super().__init__({"runId": "hybrid-run-id-123"})
+
+ @property
+ def run_id(self):
+ return self["runId"]
+
+ mock_run_pipeline.return_value = HybridModelResponse()
+
+ operator =
AzureDataFactoryRunPipelineOperator(wait_for_termination=False, **self.config)
+ operator.execute(context=self.mock_context)
+
+ assert operator.run_id == "hybrid-run-id-123"
+ self.mock_ti.xcom_push.assert_called_once_with(key="run_id",
value="hybrid-run-id-123")
+
@pytest.mark.db_test
@pytest.mark.parametrize(
("resource_group", "factory"),
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
index f7c256f04c3..419e34c00eb 100644
---
a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
@@ -135,6 +135,30 @@ class TestAzureSynapseRunSparkBatchOperator:
op.on_kill()
mock_cancel_job_run.assert_called_once_with(job_id=JOB_RUN_RESPONSE["id"])
+
@mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapseHook.get_job_run_status")
+
@mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapseHook.run_spark_job")
+ def test_job_id_extracted_from_hybrid_model_response(self,
mock_run_spark_job, mock_get_job_run_status):
+ """Regression test: azure SDK v10 hybrid models don't expose
attributes via vars()."""
+
+ class HybridModelResponse(dict):
+ """Simulates an azure SDK v10 hybrid model response."""
+
+ def __init__(self):
+ super().__init__({"id": 456})
+
+ @property
+ def id(self):
+ return self["id"]
+
+ mock_get_job_run_status.return_value = "success"
+ mock_run_spark_job.return_value = HybridModelResponse()
+
+ op = AzureSynapseRunSparkBatchOperator(
+ task_id="test", azure_synapse_conn_id=AZURE_SYNAPSE_CONN_ID,
spark_pool="test_pool", payload={}
+ )
+ op.execute(context=self.mock_context)
+ assert op.job_id == 456
+
class TestAzureSynapseRunPipelineOperator:
@pytest.fixture(autouse=True)
@@ -288,6 +312,28 @@ class TestAzureSynapseRunPipelineOperator:
# Checking the pipeline run status should _not_ be called when
``wait_for_termination`` is False.
mock_get_pipeline_run.assert_not_called()
+
@mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapsePipelineHook.run_pipeline")
+ def test_run_id_extracted_from_hybrid_model_response(self,
mock_run_pipeline):
+ """Regression test: azure SDK v10 hybrid models don't expose
attributes via vars()."""
+
+ class HybridModelResponse(dict):
+ """Simulates an azure SDK v10 hybrid model response."""
+
+ def __init__(self):
+ super().__init__({"runId": "hybrid-run-id-456"})
+
+ @property
+ def run_id(self):
+ return self["runId"]
+
+ mock_run_pipeline.return_value = HybridModelResponse()
+
+ operator = AzureSynapseRunPipelineOperator(wait_for_termination=False,
**self.config)
+ operator.execute(context=self.mock_context)
+
+ assert operator.run_id == "hybrid-run-id-456"
+ self.mock_ti.xcom_push.assert_called_once_with(key="run_id",
value="hybrid-run-id-456")
+
@pytest.mark.db_test
def test_run_pipeline_operator_link(
self,
