Fix coroutine serialization error in PowerBIDatasetRefreshOperator

[henry3260]

Why

When the PowerBIDatasetRefreshOperator is executed with wait_for_completion=False, it takes the synchronous "fire-and-forget" path. However, the underlying hook method trigger_dataset_refresh is an async function.

What

Forced Synchronous Execution: Wrapped the hook.trigger_dataset_refresh(...) call with asyncio.run(...) inside the wait_for_completion=False block. This ensures the async hook method executes fully within the current synchronous thread and returns the actual string ID.

closes: #63811

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[henry3260]

cc @morelgeorge

[SameerMesiah97]

I think you found the cause but I feel the fix might be at the wrong layer. Also, did you manage to reproduce the bug locally?

[henry3260]

I think you found the cause but I feel the fix might be at the wrong layer. Also, did you manage to reproduce the bug locally?

Thanks for reviewing! Yes, I reproduced the error locally with the unit test.

[morelgeorge]

Hello @henry3260,
thank you for your effort! I tried to run the updated provider in our test instance. Unfortunately, I still have an issue with wait_for_completion=False.
Currently I am receiving an AirflowNotFoundException connection error. See :

PowerBIDatasetRefreshException: Failed to trigger dataset refresh.
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1068 in run
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1477 in _execute_task
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 417 in wrapper
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/operators/powerbi.py", line 126 in execute
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 201 in trigger_dataset_refresh_sync
File "/appl/airflow/lib/lib/python3.12/asyncio/runners.py", line 195 in run
File "/appl/airflow/lib/lib/python3.12/asyncio/runners.py", line 118 in run
File "/appl/airflow/lib/lib/python3.12/asyncio/base_events.py", line 691 in run_until_complete
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 236 in trigger_dataset_refresh
AirflowNotFoundException: The conn_id `PowerBI_Con8` isn't defined
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 222 in trigger_dataset_refresh
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 511 in run
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 583 in send_request
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 404 in get_async_conn
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/common/compat/connection/__init__.py", line 43 in get_async_connection
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/bases/hook.py", line 75 in aget_connection
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 234 in async_get
File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 220 in _async_get_connection

The connection is defined directly in UI. In case the wait_for_completion is set to True, DAG works fine. So, it seems that once you run it synchronously through asyncio.run() then get_async_conn does not work properly.

I did a quick test to test this behaviour. I changed line 583 in msgraph hook in method send_request. Instead of conn = await self.get_async_conn() I tried to use deprecated conn = self.get_conn().

Then wait_for_completion works fine - even for True or False

[henry3260]

Hello @henry3260,

thank you for your effort! I tried to run the updated provider in our test instance. Unfortunately, I still have an issue with wait_for_completion=False.

Currently I am receiving an AirflowNotFoundException connection error. See :

PowerBIDatasetRefreshException: Failed to trigger dataset refresh.

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1068 in run

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1477 in _execute_task

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 417 in wrapper

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/operators/powerbi.py", line 126 in execute

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 201 in trigger_dataset_refresh_sync

File "/appl/airflow/lib/lib/python3.12/asyncio/runners.py", line 195 in run

File "/appl/airflow/lib/lib/python3.12/asyncio/runners.py", line 118 in run

File "/appl/airflow/lib/lib/python3.12/asyncio/base_events.py", line 691 in run_until_complete

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 236 in trigger_dataset_refresh

AirflowNotFoundException: The conn_id `PowerBI_Con8` isn't defined

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/powerbi.py", line 222 in trigger_dataset_refresh

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 511 in run

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 583 in send_request

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 404 in get_async_conn

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/providers/common/compat/connection/__init__.py", line 43 in get_async_connection

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/bases/hook.py", line 75 in aget_connection

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 234 in async_get

File "/appl/airflow/lib/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 220 in _async_get_connection

The connection is defined directly in UI. In case the wait_for_completion is set to True, DAG works fine. So, it seems that once you run it synchronously through asyncio.run() then get_async_conn does not work properly.

I did a quick test to test this behaviour. I changed line 583 in msgraph hook in method send_request. Instead of conn = await self.get_async_conn() I tried to use deprecated conn = self.get_conn().

Then wait_for_completion works fine - even for True or False

Hi @morelgeorge! Thanks for the information! I think this version is better and will avoid the above error.