Let MSGraphAsyncOperator and MSGraphSensor start from the triggerer

[dabla]

MSGraphAsyncOperator and MSGraphSensor do all their requests in the triggerer, so the worker they start on only hands the first request over. With the new start_from_trigger=True argument the scheduler defers the task itself and that first worker run is skipped.

This PR was blocked on template rendering: a task which starts from a trigger got its trigger built from unrendered fields. That is solved since #55068 (Airflow 3.3.0), where the triggerer renders the templated fields the trigger shares with the operator. The branch is rebuilt as one commit on current main, as the msgraph code changed a lot since it was opened.

What changed since the earlier version of this PR

The earlier review asked to always start from the trigger. This version makes it opt-in instead, like FileSensor, TimeSensor, DateTimeSensorAsync and DataprocSubmitJobOperator, because starting from the trigger is not transparent for existing Dags:

• the triggerer renders with a plain Jinja environment, without the user_defined_macros and user_defined_filters of the Dag, and it always renders strings, also with render_template_as_native_obj;
• the task goes straight from scheduled to deferred, so the first request does not wait for a pool slot or for a worker.

When a task still starts on a worker

With the argument set, a task falls back to the worker path without failing, so it can be set through default_args:

• on Airflow older than 3.3 (3.1 and 3.2 ignore start_from_trigger, 2.11 and 3.0 honour it but do not render templates);
• when an argument which is passed on to the trigger is an XComArg, a callable or a file-like object: the triggerer never resolves an XComArg, and the other two cannot be serialized;
• when the task is mapped: the class attributes stay at their defaults, so a mapped task has no start_trigger_args for the scheduler to use.

execute(), execute_complete() and the pagination are untouched; they remain the path for those cases and for every following page or poll. For the sensor the StartTriggerArgs carry the sensor timeout, since the scheduler only applies the timeout it is handed.

Tests

• the trigger arguments which are built, and that the trigger made from them serializes identically to the one execute() defers;
• every fallback (XComArg, also nested in a dict or list, callable, file-like object, Airflow version, mapped task);
• with a database: DagRun.schedule_tis defers the task, the Trigger row holds the arguments and, for the sensor, the task instance gets the trigger timeout, while a task which fell back is scheduled as usual;
• with a database: the trigger renders the templated fields once it gets its task instance, and sends the rendered request;
• the whole flow for both classes, through tests_common.test_utils.operators.run_deferrable, which now builds and renders the trigger from start_trigger_args when an operator starts from the trigger.

Run locally on main: the msgraph operator, sensor, trigger and notifier tests, the Power BI and Analysis Services tests and the WinRM operator tests, which share the test helper (111 passed), prek pre-commit and manual stages including mypy for providers. test_generic_transfer.py of common.sql also uses the helper, but could not be run locally (no MySQL client library to build mysqlclient), so that one is left to CI.

Known limitation, not addressed here

For a task with start_from_trigger, the triggerer renders every trigger of that task, not only the one the scheduler created. The triggers execute_complete defers for the next page or the next poll hold values which were already rendered on the worker, and pass through Jinja a second time. That only matters when a rendered value itself contains Jinja markers.

---

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

• Yes — Claude Code (Fable 5.1)

Generated-by: Claude Code (Fable 5.1) following the guidelines

---

• 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.

[dabla]

I think there is an issue with the defer_task method of TaskInstance, it should call render_template_fields first before calling the expand_start_trigger_args method. Otherwise non rendered fields would be passed to the triggerer to instantiate, which is a not the same behaviour as with regular operators defering afterwards.

[dabla]

Ok overriding the expand_start_trigger_args doesn't help because the scheduler has a serialized version of the task (e.g. operator), so it won't take into account that overriden method. So now remains the question, how do we fix this issue. This is clearly a situation which hasn't been thought about.

[dabla]

This PR must be merged before this one can.

[dabla]

Created new PR which attempts to fix template rendering with start from trigger, as long as template rendering in triggerer isn't fixed, this PR can't be merged.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.