Add wait_for_termination parameter and fix double-deferral in PowerBIDatasetRefreshOperator

[shreyas-dev]

Description

This PR adds an optional wait_for_termination parameter to PowerBIDatasetRefreshOperator and refactors the execution flow to fix the double-deferral anti-pattern present in the current implementation.

Motivation and Context

New Feature:
Users need the ability to trigger Power BI dataset refreshes without waiting for completion. This is useful for:

• Long-running refreshes (hours/days) that don't need to block the DAG
• Scenarios where downstream tasks don't depend on refresh completion
• Reducing worker slot usage for fire-and-forget operations

Bug Fix:
The current implementation uses an inefficient double-deferral pattern:

1. execute() → defer → triggers refresh and returns dataset_refresh_id
2. get_refresh_status() → defer again → polls for completion
3. execute_complete() → processes final result

This is unnecessary because PowerBITrigger.run() already handles both triggering the refresh AND polling for completion in a single async operation when wait_for_termination=True.

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

• No

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[dabla]

Congratulations @shreyas-dev, looking good to me beside the remark made by @shahar1 I think where are good nice improvement.

[shreyas-dev]

Hey @dabla and @shahar1, Added wait_for_termination to docstring, please review.

[shreyas-dev]

Hey @dabla, fixed "imperative mood" docstring test case which was failing.

[shreyas-dev]

Hey @vincbeck / @dabla, Please check

[shreyas-dev]

Hey @vincbeck , Resolved www-hash.txt.

[shreyas-dev]

Hey @vincbeck,

_____________________________________________________________________________ TestAwsS3Hook.test_sync_to_local_dir_behaviour _____________________________________________________________________________
[gw1] linux -- Python 3.10.19 /usr/python/bin/python
providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py:1840: in test_sync_to_local_dir_behaviour
    assert "S3 object size" in logs_string
E   AssertionError: assert 'S3 object size' in 'Downloading data from s3://airflow-test-s3-bucket/ to /tmp/pytest-of-root/pytest-0/popen-gw1/test_sync_to_local_dir_b...al_dir_behavio0/s3_sync_dir/subproject1/dag_b.py is up-to-date with S3 object subproject1/dag_b.py. Skipping download.'

This test case failure looks unrelated to this PR. I have updated my branch. Can you please check ?

[shreyas-dev]

@dabla , Let me know if anything else is needed to get this merged.

[shreyas-dev]

Used structlog instead of logging.

[shreyas-dev]

Ran prek --all-files locally and all results passed

[shreyas-dev]

Hey @vincbeck , I'm guessing structlog was introduced in airflow 3+ and airflow 2.11 used standard logging. How to resolve this issue? On the contrary I'm now thinking of removing the warning cause Its a just informational message.

logger.warning(
                "The PowerBIDatasetRefreshOperator now always uses deferrable execution when wait_for_termination=True."
            )

In this case I can remove structlog entirely

[vincbeck]

Hey @vincbeck , I'm guessing structlog was introduced in airflow 3+ and airflow 2.11 used standard logging. How to resolve this issue? On the contrary I'm now thinking of removing the warning cause Its a just informational message.

logger.warning(
                "The PowerBIDatasetRefreshOperator now always uses deferrable execution when wait_for_termination=True."
            )

In this case I can remove structlog entirely

Can you check how it is done in the other providers?

[shreyas-dev]

Hey @vincbeck, Updated the code.prek --all-files was successful locally. Please check

[shreyas-dev]

The failed test doesn't look to be related to this PR https://github.com/apache/airflow/actions/runs/21616745897/job/62362457618?pr=60369

Host python version is 3.10.19 (main, Oct 10 2025, 01:02:36) [GCC 11.4.0]
The image /mnt/ci-image-save-v3-linux_amd64-3.10.tar does not exist.
Error: Process completed with exit code 1.

[shreyas-dev]

Hey @vincbeck, Apologies I have fixed the test case and verified all the test cases by running it locally.

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.