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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
