From bb301ad3f0b296d7e74fde8e6d4230449547b6b7 Mon Sep 17 00:00:00 2001 From: Malte Niederstadt Date: Fri, 10 Jul 2026 07:36:53 +0200 Subject: [PATCH 1/3] fix(providers/azure): use attribute access instead of vars() for SDK v10 hybrid model compat --- .../providers/microsoft/azure/operators/data_factory.py | 2 +- .../airflow/providers/microsoft/azure/operators/synapse.py | 4 ++-- 2 files changed, 3 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 495eb27d6a7f6..36321dbe10a28 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 @@ def execute(self, context: Context) -> None: 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 b0c171a13bd53..19a082ced77d6 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 @@ def execute(self, context: Context) -> None: 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 @@ def execute(self, context) -> None: 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. From 9156800206e338e71064634eaf4b43dd11b0080c Mon Sep 17 00:00:00 2001 From: Malte Niederstadt Date: Fri, 10 Jul 2026 09:29:35 +0200 Subject: [PATCH 2/3] added regression tests for azure data factory and azure synapse operators --- .../azure/operators/test_data_factory.py | 26 ++++++++++ .../microsoft/azure/operators/test_synapse.py | 49 +++++++++++++++++++ 2 files changed, 75 insertions(+) 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 c640303048061..7804891fd4b06 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 @@ def test_execute_no_wait_for_termination(self, mock_run_pipeline): # 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 f7c256f04c35b..255eb1e2de739 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 @@ -136,6 +136,31 @@ def test_azure_synapse_run_spark_batch_operator_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) def setup_test_cases(self, create_mock_connection): @@ -288,6 +313,30 @@ def test_execute_no_wait_for_termination(self, mock_run_pipeline): # 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, From 1f5be5486d7e52dd6884ba59235e3c1f70b3a8b7 Mon Sep 17 00:00:00 2001 From: Malte Niederstadt Date: Fri, 10 Jul 2026 10:30:46 +0200 Subject: [PATCH 3/3] Fix ruff formatting in test_synapse.py --- .../tests/unit/microsoft/azure/operators/test_synapse.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) 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 255eb1e2de739..419e34c00eb50 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,7 +135,6 @@ def test_azure_synapse_run_spark_batch_operator_on_kill( 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): @@ -313,9 +312,7 @@ def test_execute_no_wait_for_termination(self, mock_run_pipeline): # 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" - ) + @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()."""