You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dabla (via GitHub)" <gi...@apache.org> on 2025/07/08 09:14:52 UTC

[PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

dabla opened a new pull request, #53009:
URL: https://github.com/apache/airflow/pull/53009

   <!--
    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at
   
      http://www.apache.org/licenses/LICENSE-2.0
   
    Unless required by applicable law or agreed to in writing,
    software distributed under the License is distributed on an
    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    KIND, either express or implied.  See the License for the
    specific language governing permissions and limitations
    under the License.
    -->
   
   <!--
   Thank you for contributing! Please make sure that your code changes
   are covered with tests. And in case of new features or big changes
   remember to adjust the documentation.
   
   Feel free to ping committers for the review!
   
   In case of an existing issue, reference it using one of the following:
   
   closes: #ISSUE
   related: #ISSUE
   
   How to write a good git commit message:
   http://chris.beams.io/posts/git-commit/
   -->
   
   Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker a this avoids unnecessary round trips to the scheduler.
   
   <!-- Please keep an empty line above the dashes. -->
   ---
   **^ Add meaningful description above**
   Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
   In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
   In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
   In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r2192165544


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -108,8 +125,17 @@ class MSGraphAsyncOperator(BaseOperator):
         the message from the event, otherwise the response from the event payload is returned.
     :param serializer: Class which handles response serialization (default is ResponseSerializer).
         Bytes will be base64 encoded into a string, so it can be stored as an XCom.
+    :param start_from_trigger: If set to True, the operator will start directly from the triggerer without going into the worker first.
     """
 
+    start_trigger_args = StartTriggerArgs(
+        trigger_cls=f"{MSGraphTrigger.__module__}.{MSGraphTrigger.__name__}",
+        trigger_kwargs={},
+        next_method="execute_complete",
+        next_kwargs=None,
+        timeout=None,
+    )
+    start_from_trigger = False

Review Comment:
   Is there a reason you we wouldn't want this to be true by default? Given all the execute does is defer, I don't even think this needs to be an option -- just set it to true always? And then `execute()` is only called for when running on older versions of Airflow



##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -35,6 +37,21 @@
 from airflow.providers.microsoft.azure.version_compat import BaseOperator
 from airflow.utils.xcom import XCOM_RETURN_KEY
 
+try:
+    from airflow.triggers.base import StartTriggerArgs
+except ImportError:
+    # TODO: Remove this when min airflow version is 2.10.0 for standard provider

Review Comment:
   ```suggestion
       # TODO: Remove this when min airflow version is 2.10.0
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #53009: Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker
URL: https://github.com/apache/airflow/pull/53009


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-3051428503

   This [PR ](https://github.com/apache/airflow/pull/53071) must be merged before this one can.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r2192610832


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -108,8 +125,17 @@ class MSGraphAsyncOperator(BaseOperator):
         the message from the event, otherwise the response from the event payload is returned.
     :param serializer: Class which handles response serialization (default is ResponseSerializer).
         Bytes will be base64 encoded into a string, so it can be stored as an XCom.
+    :param start_from_trigger: If set to True, the operator will start directly from the triggerer without going into the worker first.
     """
 
+    start_trigger_args = StartTriggerArgs(
+        trigger_cls=f"{MSGraphTrigger.__module__}.{MSGraphTrigger.__name__}",
+        trigger_kwargs={},
+        next_method="execute_complete",
+        next_kwargs=None,
+        timeout=None,
+    )
+    start_from_trigger = False

Review Comment:
   I've refactored code to always run from triggerer and modified tests to support this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-3049859828

   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.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r2192489101


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -108,8 +125,17 @@ class MSGraphAsyncOperator(BaseOperator):
         the message from the event, otherwise the response from the event payload is returned.
     :param serializer: Class which handles response serialization (default is ResponseSerializer).
         Bytes will be base64 encoded into a string, so it can be stored as an XCom.
+    :param start_from_trigger: If set to True, the operator will start directly from the triggerer without going into the worker first.
     """
 
+    start_trigger_args = StartTriggerArgs(
+        trigger_cls=f"{MSGraphTrigger.__module__}.{MSGraphTrigger.__name__}",
+        trigger_kwargs={},
+        next_method="execute_complete",
+        next_kwargs=None,
+        timeout=None,
+    )
+    start_from_trigger = False

Review Comment:
   Yes indeed, fully agree, but I did this to allow the tests to run, which  is a bad reason but would need some rework



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-3049784739

   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 ](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/models/taskinstance.py#L1607) method.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Let MSGraphAsyncOperator and MSGraphSensor start from the triggerer [airflow]

Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-6489842769

   @dabla Could you please resolve conflicts?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r2192491425


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -35,6 +37,21 @@
 from airflow.providers.microsoft.azure.version_compat import BaseOperator
 from airflow.utils.xcom import XCOM_RETURN_KEY
 
+try:
+    from airflow.triggers.base import StartTriggerArgs
+except ImportError:
+    # TODO: Remove this when min airflow version is 2.10.0 for standard provider

Review Comment:
   copy/pasta error :)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-4043055199

   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.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #53009:
URL: https://github.com/apache/airflow/pull/53009#issuecomment-3246843750

   Created new [PR](https://github.com/apache/airflow/pull/55068) 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.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Allow MSGraphAsyncOperator and MSGraphSensor to directly start from trigger instead of worker [airflow]

Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r2192525827


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -108,8 +125,17 @@ class MSGraphAsyncOperator(BaseOperator):
         the message from the event, otherwise the response from the event payload is returned.
     :param serializer: Class which handles response serialization (default is ResponseSerializer).
         Bytes will be base64 encoded into a string, so it can be stored as an XCom.
+    :param start_from_trigger: If set to True, the operator will start directly from the triggerer without going into the worker first.
     """
 
+    start_trigger_args = StartTriggerArgs(
+        trigger_cls=f"{MSGraphTrigger.__module__}.{MSGraphTrigger.__name__}",
+        trigger_kwargs={},
+        next_method="execute_complete",
+        next_kwargs=None,
+        timeout=None,
+    )
+    start_from_trigger = False

Review Comment:
   I'll try to refactor the tests to support this aspect only, it would then indeed make more sense to always start the operator from the triggerer



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Let MSGraphAsyncOperator and MSGraphSensor start from the triggerer [airflow]

Posted by "shahar1 (via GitHub) via commits" <co...@airflow.apache.org>.
shahar1 commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r4500398457


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
     MSGraphTrigger,
     ResponseSerializer,
 )
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
+    from datetime import timedelta
     from io import BytesIO
 
     from msgraph_core import APIVersion
 
+    from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
     from airflow.sdk import Context
 
 
+def _is_plain_data(value: Any) -> bool:
+    if isinstance(value, dict):
+        return all(isinstance(key, str) and _is_plain_data(item) for key, item in value.items())
+    if isinstance(value, (list, tuple)):
+        return all(_is_plain_data(item) for item in value)
+    return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+    operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None = None
+) -> StartTriggerArgs | None:
+    """
+    Build the arguments which start the given operator from the triggerer, when that is possible.
+
+    The triggerer renders the templated fields of the trigger itself, but only since Airflow 3.3,
+    and it only gets what the serialized Dag holds.  It never resolves an ``XComArg``, and a
+    callable or a file-like object cannot be serialized.  None is returned in those cases, so the
+    operator starts on a worker instead.
+
+    :param operator: The operator to build the arguments for.
+    :param timeout: How long the trigger may run before the task times out (default is None).
+    """
+    if not AIRFLOW_V_3_3_PLUS:
+        return None
+
+    serializer = type(operator.serializer)
+    trigger_kwargs = {
+        "url": operator.url,
+        "response_type": operator.response_type,
+        "path_parameters": operator.path_parameters,
+        "url_template": operator.url_template,
+        "method": operator.method,
+        "query_parameters": operator.query_parameters,
+        "headers": operator.headers,
+        "data": operator.data,
+        "conn_id": operator.conn_id,
+        "timeout": operator.timeout,

Review Comment:
   **Minor: the sensor timeout should not reach the trigger any more.**
   
   #75774 removed `timeout=self.timeout` from the trigger the sensor defers, as the sensor timeout is not an HTTP timeout. This shared helper still passes it, so the first poll would use it as the HTTP timeout and the later polls none. `test_start_from_trigger_defers_the_same_trigger_as_execute` for the sensor should fail on this after the rebase. The fix belongs here, not in that test.



##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
     MSGraphTrigger,
     ResponseSerializer,
 )
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
+    from datetime import timedelta
     from io import BytesIO
 
     from msgraph_core import APIVersion
 
+    from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
     from airflow.sdk import Context
 
 
+def _is_plain_data(value: Any) -> bool:
+    if isinstance(value, dict):
+        return all(isinstance(key, str) and _is_plain_data(item) for key, item in value.items())
+    if isinstance(value, (list, tuple)):
+        return all(_is_plain_data(item) for item in value)
+    return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+    operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None = None
+) -> StartTriggerArgs | None:
+    """
+    Build the arguments which start the given operator from the triggerer, when that is possible.
+
+    The triggerer renders the templated fields of the trigger itself, but only since Airflow 3.3,
+    and it only gets what the serialized Dag holds.  It never resolves an ``XComArg``, and a
+    callable or a file-like object cannot be serialized.  None is returned in those cases, so the
+    operator starts on a worker instead.
+
+    :param operator: The operator to build the arguments for.
+    :param timeout: How long the trigger may run before the task times out (default is None).
+    """
+    if not AIRFLOW_V_3_3_PLUS:

Review Comment:
   **Major: on Airflow 3.3.x a retry or a clear reads the XComs of the earlier attempt.**
   
   `ti_run` only asks the worker to clear XComs under `if not ti.next_method:`, and a task coming from `defer_task` already has `next_method` set on its first worker run. On 3.3.x the XCom rows are keyed by Dag, task, run and map index, and nothing else removes them on a retry or a clear. `execute_complete` then starts from `results = self.pull_xcom(context=context)`, so:
   
   - a retry after a failure on page 2 returns page 1 of the earlier attempt plus page 1 again;
   - clearing a finished task reads its earlier return value. For a dict response, `list(value)` gives its keys, and the task returns `[*keys, response]`.
   
   3.4 keys XComs per attempt and rotates the task instance id, so this only affects 3.3.x, and only the operator: the sensor does not pull XComs.
   
   The smallest fix is to raise this gate to 3.4. If 3.3 support matters, `execute_complete` needs to know it is handling the first page, for instance through a marker in the `defer_to_next_link` kwargs, and start from an empty list then. Either way, please add a test.



##########
providers/microsoft/azure/docs/operators/msgraph.rst:
##########
@@ -80,6 +80,40 @@ Below is an example of using this operator to create an item schedule in Fabric.
     :start-after: [START howto_operator_ms_fabric_create_item_schedule]
     :end-before: [END howto_operator_ms_fabric_create_item_schedule]
 
+.. _howto/operator:MSGraphAsyncOperator:start_from_trigger:
+
+Starting from the triggerer
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The operator does all its requests in the triggerer, so the only thing it does on a worker at the start
+is to hand the first request over. With ``start_from_trigger=True`` the scheduler defers the task itself
+and that first worker run is skipped. The worker gets involved once the response is there, to process it
+and to follow the pagination.
+
+.. code-block:: python
+
+    site_task = MSGraphAsyncOperator(
+        task_id="news_site",
+        conn_id="msgraph_api",
+        url="sites/{{ params.site }}",
+        start_from_trigger=True,
+    )
+
+This needs Airflow 3.3 or later, on older versions the task starts on a worker as it does without the
+argument. It also starts on a worker when:
+
+* an argument which is passed on to the trigger is an ``XComArg``, a callable or a file-like object,
+  as the triggerer cannot resolve or receive those;
+* the task is mapped with ``expand``.
+
+Keep the following in mind when you enable it:

Review Comment:
   **Minor: the second render pass is only described in the PR description.**
   
   The triggerer takes `template_fields` from `task.start_trigger_args` for every trigger of the task, so the triggers `execute_complete` and `retry_execute` defer are rendered again. That covers values already rendered on the worker, and also the `nextLink` and query parameters which come from a Graph response. A value holding Jinja markers then changes the request or fails it.
   
   I would rather fix this than leave it: `MSGraphTrigger` could take a flag for triggers built from rendered values, with `render_template_fields` doing nothing when it is set. If it stays, users need to read about it in this list.



##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -161,6 +224,8 @@ def __init__(
         self.result_processor = result_processor
         self.event_handler = event_handler or default_event_handler
         self.serializer: ResponseSerializer = serializer()
+        self.start_trigger_args = build_start_trigger_args(self) if start_from_trigger else None

Review Comment:
   **Major: `durable=True` loses its crash recovery here (needs handling in the rebase).**
   
   Since #74675, `execute()` is where a durable task reads its state: it returns the kept result of a completed non-GET request, and it resumes from the cursor. A start-from-trigger retry never calls `execute()`, because `DagRun.schedule_tis` goes to `TaskInstance.defer_task`, which creates the first-page trigger again. With `durable=True, start_from_trigger=True`, a retry after a crash then:
   
   - sends a non-GET request again even though it completed, which is what `_COMPLETED_KEY` exists to prevent;
   - fetches page 1 again, and `push_page` stores it at `cursor["pages"]`, after the pages already kept, so the final result holds those pages twice.
   
   Suggested fix: a durable task starts on a worker, like the other fallbacks. Please add it to the docs and cover it with a test.
   
   ```python
   self.start_trigger_args = (
       build_start_trigger_args(self) if start_from_trigger and not durable else None
   )
   ```



##########
providers/microsoft/azure/docs/sensors/msgraph.rst:
##########
@@ -33,6 +33,24 @@ Below is an example of using this sensor to poll the status of a PowerBI workspa
     :start-after: [START howto_sensor_powerbi_scan_status]
     :end-before: [END howto_sensor_powerbi_scan_status]
 
+Starting from the triggerer
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+With ``start_from_trigger=True`` the scheduler defers the sensor itself, so the first poll is done without
+a worker being involved. The ``timeout`` of the sensor also applies to that first poll. The requirements

Review Comment:
   **Minor: the timeout budget restarts after the first poll.**
   
   #75774 carries `sensor_started_at` through the trigger kwargs so re-deferring does not reset the timeout. `StartTriggerArgs` has no such value, because it is built at parse time, so `execute_complete` gets `None` and `_started_at` uses the current time. The first poll can use the whole `timeout` under `trigger_timeout`, and the budget then starts over, so the total can reach about twice `timeout`. Either record when the first poll started, or say this here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] Let MSGraphAsyncOperator and MSGraphSensor start from the triggerer [airflow]

Posted by "dabla (via GitHub) via commits" <co...@airflow.apache.org>.
dabla commented on code in PR #53009:
URL: https://github.com/apache/airflow/pull/53009#discussion_r4507384528


##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -161,6 +224,8 @@ def __init__(
         self.result_processor = result_processor
         self.event_handler = event_handler or default_event_handler
         self.serializer: ResponseSerializer = serializer()
+        self.start_trigger_args = build_start_trigger_args(self) if start_from_trigger else None

Review Comment:
   Good catch. A durable task now always starts on a worker: `start_trigger_args` is only built when `start_from_trigger and not durable`, as you suggested. The docs list `durable=True` with the other fallbacks, and `test_durable_task_starts_on_a_worker` covers it. Fixed in 8394dfbfcc.



##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
     MSGraphTrigger,
     ResponseSerializer,
 )
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
+    from datetime import timedelta
     from io import BytesIO
 
     from msgraph_core import APIVersion
 
+    from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
     from airflow.sdk import Context
 
 
+def _is_plain_data(value: Any) -> bool:
+    if isinstance(value, dict):
+        return all(isinstance(key, str) and _is_plain_data(item) for key, item in value.items())
+    if isinstance(value, (list, tuple)):
+        return all(_is_plain_data(item) for item in value)
+    return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+    operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None = None
+) -> StartTriggerArgs | None:
+    """
+    Build the arguments which start the given operator from the triggerer, when that is possible.
+
+    The triggerer renders the templated fields of the trigger itself, but only since Airflow 3.3,
+    and it only gets what the serialized Dag holds.  It never resolves an ``XComArg``, and a
+    callable or a file-like object cannot be serialized.  None is returned in those cases, so the
+    operator starts on a worker instead.
+
+    :param operator: The operator to build the arguments for.
+    :param timeout: How long the trigger may run before the task times out (default is None).
+    """
+    if not AIRFLOW_V_3_3_PLUS:

Review Comment:
   Confirmed: on 3.3, `ti_run` only clears XComs when `next_method` is unset, and a task the scheduler deferred always has it. I went with the smaller fix: starting from the triggerer now needs Airflow 3.4 (`AIRFLOW_V_3_4_PLUS`), and on 3.3 the task starts on a worker. Keeping 3.3 would have needed a first-page marker that also changes the worker path, which is more code for a version that already needed 3.3.2 for the triggerer crash loop. Docs and docstrings say 3.4, and `test_start_from_trigger_needs_airflow_3_4` covers the fallback. Fixed in 8394dfbfcc.



##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/msgraph.py:
##########
@@ -32,15 +32,73 @@
     MSGraphTrigger,
     ResponseSerializer,
 )
+from airflow.providers.microsoft.azure.version_compat import AIRFLOW_V_3_3_PLUS
+from airflow.triggers.base import StartTriggerArgs
 
 if TYPE_CHECKING:
+    from datetime import timedelta
     from io import BytesIO
 
     from msgraph_core import APIVersion
 
+    from airflow.providers.microsoft.azure.sensors.msgraph import MSGraphSensor
     from airflow.sdk import Context
 
 
+def _is_plain_data(value: Any) -> bool:
+    if isinstance(value, dict):
+        return all(isinstance(key, str) and _is_plain_data(item) for key, item in value.items())
+    if isinstance(value, (list, tuple)):
+        return all(_is_plain_data(item) for item in value)
+    return value is None or isinstance(value, (str, int, float, bool))
+
+
+def build_start_trigger_args(
+    operator: MSGraphAsyncOperator | MSGraphSensor, timeout: timedelta | None = None
+) -> StartTriggerArgs | None:
+    """
+    Build the arguments which start the given operator from the triggerer, when that is possible.
+
+    The triggerer renders the templated fields of the trigger itself, but only since Airflow 3.3,
+    and it only gets what the serialized Dag holds.  It never resolves an ``XComArg``, and a
+    callable or a file-like object cannot be serialized.  None is returned in those cases, so the
+    operator starts on a worker instead.
+
+    :param operator: The operator to build the arguments for.
+    :param timeout: How long the trigger may run before the task times out (default is None).
+    """
+    if not AIRFLOW_V_3_3_PLUS:
+        return None
+
+    serializer = type(operator.serializer)
+    trigger_kwargs = {
+        "url": operator.url,
+        "response_type": operator.response_type,
+        "path_parameters": operator.path_parameters,
+        "url_template": operator.url_template,
+        "method": operator.method,
+        "query_parameters": operator.query_parameters,
+        "headers": operator.headers,
+        "data": operator.data,
+        "conn_id": operator.conn_id,
+        "timeout": operator.timeout,

Review Comment:
   You were right, that test failed after the rebase. Rather than dropping the key, I removed the hand-kept copy of the trigger arguments, which is where the drift came from. The operator and the sensor now build their first trigger in one `_build_trigger()` method each. `execute` defers it, and `build_start_trigger_args` takes `trigger.serialize()`. The start trigger is now the trigger the worker path would defer, so a later change to the trigger arguments cannot diverge again (DRY: one representation of that knowledge). The sensor's start kwargs no longer carry `timeout`. Fixed in 8394dfbfcc.



##########
providers/microsoft/azure/docs/sensors/msgraph.rst:
##########
@@ -33,6 +33,24 @@ Below is an example of using this sensor to poll the status of a PowerBI workspa
     :start-after: [START howto_sensor_powerbi_scan_status]
     :end-before: [END howto_sensor_powerbi_scan_status]
 
+Starting from the triggerer
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+With ``start_from_trigger=True`` the scheduler defers the sensor itself, so the first poll is done without
+a worker being involved. The ``timeout`` of the sensor also applies to that first poll. The requirements

Review Comment:
   Fixed instead of documented. When the first event arrives without `sensor_started_at`, the sensor now counts from `ti.start_date`. `defer_task` sets it when the scheduler defers the task, and `ti_run` keeps it when the task resumes. The total stays within `timeout`, and the docs explain how it counts. `test_first_poll_from_the_triggerer_counts_from_the_start_of_the_try` covers it. Fixed in 8394dfbfcc.



##########
providers/microsoft/azure/docs/operators/msgraph.rst:
##########
@@ -80,6 +80,40 @@ Below is an example of using this operator to create an item schedule in Fabric.
     :start-after: [START howto_operator_ms_fabric_create_item_schedule]
     :end-before: [END howto_operator_ms_fabric_create_item_schedule]
 
+.. _howto/operator:MSGraphAsyncOperator:start_from_trigger:
+
+Starting from the triggerer
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The operator does all its requests in the triggerer, so the only thing it does on a worker at the start
+is to hand the first request over. With ``start_from_trigger=True`` the scheduler defers the task itself
+and that first worker run is skipped. The worker gets involved once the response is there, to process it
+and to follow the pagination.
+
+.. code-block:: python
+
+    site_task = MSGraphAsyncOperator(
+        task_id="news_site",
+        conn_id="msgraph_api",
+        url="sites/{{ params.site }}",
+        start_from_trigger=True,
+    )
+
+This needs Airflow 3.3 or later, on older versions the task starts on a worker as it does without the
+argument. It also starts on a worker when:
+
+* an argument which is passed on to the trigger is an ``XComArg``, a callable or a file-like object,
+  as the triggerer cannot resolve or receive those;
+* the task is mapped with ``expand``.
+
+Keep the following in mind when you enable it:

Review Comment:
   Agreed, fixed rather than documented, along the lines you suggested. `MSGraphTrigger` takes a `rendered` flag (default `False`, so triggers already serialized keep their behaviour). When it is set, `render_template_fields` does nothing. Every trigger the worker defers sets it: the first request, the pagination links and the sensor polls. Only the start trigger, whose fields the triggerer must render, keeps `False`. A `nextLink` or a query value from a Graph response that contains `{{` now goes out unchanged. Covered by `TestMSGraphTriggerRendering` and `test_triggers_deferred_by_the_worker_are_not_rendered_again`. Fixed in 8394dfbfcc.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org