Fix coroutine serialization error in PowerBIDatasetRefreshOperator

Author: henry3260

providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/powerbi.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from enum import Enum
 from typing import TYPE_CHECKING, Any
 
@@ -189,6 +190,22 @@ async def get_refresh_details_by_refresh_id(
 
         return refresh_details
 
+    def trigger_dataset_refresh_sync(
+        self, *, dataset_id: str, group_id: str, request_body: dict[str, Any] | None = None
+    ) -> str:
+        """
+        Trigger Power BI Dataset refresh synchronously.
+
+        This is a wrapper around the async `trigger_dataset_refresh` method.
+        """
+        return asyncio.run(
+            self.trigger_dataset_refresh(
+                dataset_id=dataset_id,
+                group_id=group_id,
+                request_body=request_body,
+            )
+        )
+
     async def trigger_dataset_refresh(
         self, *, dataset_id: str, group_id: str, request_body: dict[str, Any] | None = None
     ) -> str:

providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/powerbi.py

@@ -123,7 +123,7 @@ def execute(self, context: Context):
                 conn_id=self.conn_id, proxies=self.proxies, api_version=self.api_version, timeout=self.timeout
             )
 
-            dataset_refresh_id = hook.trigger_dataset_refresh(
+            dataset_refresh_id = hook.trigger_dataset_refresh_sync(
                 dataset_id=self.dataset_id,
                 group_id=self.group_id,
                 request_body=self.request_body,

providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_powerbi.py

@@ -237,7 +237,7 @@ def test_powerbi_link(self, dag_maker, create_task_instance_of_operator):
     def test_execute_fire_and_forget_mode(self, mock_connection, mock_hook_class):
         """Test fire-and-forget mode (wait_for_completion=False)"""
         mock_hook_instance = mock_hook_class.return_value
-        mock_hook_instance.trigger_dataset_refresh.return_value = NEW_REFRESH_REQUEST_ID
+        mock_hook_instance.trigger_dataset_refresh_sync.return_value = NEW_REFRESH_REQUEST_ID
 
         operator = PowerBIDatasetRefreshOperator(
             **CONFIG,
@@ -250,7 +250,7 @@ def test_execute_fire_and_forget_mode(self, mock_connection, mock_hook_class):
         result = operator.execute(context)
 
         # Verify hook was called correctly
-        mock_hook_instance.trigger_dataset_refresh.assert_called_once_with(
+        mock_hook_instance.trigger_dataset_refresh_sync.assert_called_once_with(
             dataset_id=DATASET_ID,
             group_id=GROUP_ID,
             request_body=REQUEST_BODY,
@@ -261,6 +261,7 @@ def test_execute_fire_and_forget_mode(self, mock_connection, mock_hook_class):
         call_args = context["ti"].xcom_push.call_args
         assert call_args[1]["key"] == f"{TASK_ID}.powerbi_dataset_refresh_id"
         assert call_args[1]["value"] == NEW_REFRESH_REQUEST_ID
+        assert isinstance(call_args[1]["value"], str)
 
         # Should return None (completes synchronously)
         assert result is None
@@ -270,7 +271,7 @@ def test_execute_fire_and_forget_mode(self, mock_connection, mock_hook_class):
     def test_execute_fire_and_forget_mode_failure(self, mock_connection, mock_hook_class):
         """Test fire-and-forget mode raises exception when trigger fails"""
         mock_hook_instance = mock_hook_class.return_value
-        mock_hook_instance.trigger_dataset_refresh.return_value = None
+        mock_hook_instance.trigger_dataset_refresh_sync.return_value = None
 
         operator = PowerBIDatasetRefreshOperator(
             **CONFIG,