This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 16b4e621711 Give each task its own start trigger arguments in standard
sensors (#70888)
16b4e621711 is described below
commit 16b4e621711171cc8e82f3200500dcd4da1ce5d8
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat Aug 1 22:47:12 2026 +0200
Give each task its own start trigger arguments in standard sensors (#70888)
start_trigger_args is a class attribute, so assigning through it in
__init__ overwrote the arguments shared by every other task built from
the same operator. Two DateTimeSensorAsync tasks in one Dag both waited
for whichever moment was constructed last, and the same applied to
TimeSensor's moment and FileSensor's path and timeout.
---
.../providers/standard/sensors/date_time.py | 13 +++++++--
.../providers/standard/sensors/filesystem.py | 17 ++++++++----
.../src/airflow/providers/standard/sensors/time.py | 9 ++++--
.../tests/unit/standard/sensors/test_date_time.py | 32 ++++++++++++++++++++++
.../tests/unit/standard/sensors/test_filesystem.py | 25 +++++++++++++++++
.../tests/unit/standard/sensors/test_time.py | 19 +++++++++++++
6 files changed, 105 insertions(+), 10 deletions(-)
diff --git
a/providers/standard/src/airflow/providers/standard/sensors/date_time.py
b/providers/standard/src/airflow/providers/standard/sensors/date_time.py
index d000ef1cd8b..90d358c6c39 100644
--- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py
+++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import dataclasses
import datetime
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any, NoReturn
@@ -115,9 +116,15 @@ class DateTimeSensorAsync(DateTimeSensor):
self.start_from_trigger = start_from_trigger
if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self._moment,
- end_from_trigger=self.end_from_trigger,
+ # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
+ # assigning through it would overwrite the arguments of every
other task built
+ # from this operator.
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ trigger_kwargs=dict(
+ moment=self._moment,
+ end_from_trigger=self.end_from_trigger,
+ ),
)
def execute(self, context: Context) -> NoReturn:
diff --git
a/providers/standard/src/airflow/providers/standard/sensors/filesystem.py
b/providers/standard/src/airflow/providers/standard/sensors/filesystem.py
index 817ba10b3a4..307c6d901e8 100644
--- a/providers/standard/src/airflow/providers/standard/sensors/filesystem.py
+++ b/providers/standard/src/airflow/providers/standard/sensors/filesystem.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import dataclasses
import datetime
import os
from collections.abc import Sequence
@@ -90,11 +91,17 @@ class FileSensor(BaseSensorOperator):
self.start_from_trigger = start_from_trigger
if self.deferrable and self.start_from_trigger:
- self.start_trigger_args.timeout =
datetime.timedelta(seconds=self.timeout)
- self.start_trigger_args.trigger_kwargs = dict(
- filepath=self.path,
- recursive=self.recursive,
- poke_interval=self.poke_interval,
+ # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
+ # assigning through it would overwrite the arguments of every
other task built
+ # from this operator.
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ timeout=datetime.timedelta(seconds=self.timeout),
+ trigger_kwargs=dict(
+ filepath=self.path,
+ recursive=self.recursive,
+ poke_interval=self.poke_interval,
+ ),
)
@cached_property
diff --git a/providers/standard/src/airflow/providers/standard/sensors/time.py
b/providers/standard/src/airflow/providers/standard/sensors/time.py
index e32a98d97d9..2e3b19eb899 100644
--- a/providers/standard/src/airflow/providers/standard/sensors/time.py
+++ b/providers/standard/src/airflow/providers/standard/sensors/time.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import dataclasses
import datetime
import warnings
from typing import TYPE_CHECKING, Any
@@ -81,8 +82,12 @@ class TimeSensor(BaseSensorOperator):
self.end_from_trigger = end_from_trigger
if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self.target_datetime,
end_from_trigger=self.end_from_trigger
+ # Replaced rather than mutated: ``start_trigger_args`` is a class
attribute, so
+ # assigning through it would overwrite the arguments of every
other task built
+ # from this operator.
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ trigger_kwargs=dict(moment=self.target_datetime,
end_from_trigger=self.end_from_trigger),
)
def execute(self, context: Context) -> None:
diff --git a/providers/standard/tests/unit/standard/sensors/test_date_time.py
b/providers/standard/tests/unit/standard/sensors/test_date_time.py
index 96d125750ce..e55569e63c2 100644
--- a/providers/standard/tests/unit/standard/sensors/test_date_time.py
+++ b/providers/standard/tests/unit/standard/sensors/test_date_time.py
@@ -157,3 +157,35 @@ class TestDateTimeSensor:
dag=self.dag,
)
assert op.start_trigger_args.trigger_kwargs["moment"] ==
pendulum.datetime(2020, 1, 1, tz="UTC")
+
+ def test_start_trigger_args_are_not_shared_between_tasks(self):
+ """Each task must carry its own trigger arguments.
+
+ ``start_trigger_args`` is a class attribute, so assigning through it
made every task
+ built from this operator advertise the moment of whichever was
constructed last.
+ """
+ first = DateTimeSensorAsync(
+ task_id="first",
+ target_time="2030-01-01T00:00:00+00:00",
+ start_from_trigger=True,
+ dag=self.dag,
+ )
+ second = DateTimeSensorAsync(
+ task_id="second",
+ target_time="2040-06-06T00:00:00+00:00",
+ start_from_trigger=True,
+ dag=self.dag,
+ )
+
+ assert first.start_trigger_args is not second.start_trigger_args
+ assert first.start_trigger_args.trigger_kwargs["moment"] ==
pendulum.parse(
+ "2030-01-01T00:00:00+00:00"
+ )
+ assert second.start_trigger_args.trigger_kwargs["moment"] ==
pendulum.parse(
+ "2040-06-06T00:00:00+00:00"
+ )
+ # the class level template must survive untouched for the next task
built from it
+ assert DateTimeSensorAsync.start_trigger_args.trigger_kwargs == {
+ "moment": "",
+ "end_from_trigger": False,
+ }
diff --git a/providers/standard/tests/unit/standard/sensors/test_filesystem.py
b/providers/standard/tests/unit/standard/sensors/test_filesystem.py
index dc94f3d6c03..a203b379572 100644
--- a/providers/standard/tests/unit/standard/sensors/test_filesystem.py
+++ b/providers/standard/tests/unit/standard/sensors/test_filesystem.py
@@ -238,3 +238,28 @@ class TestFileSensor:
task.execute({})
assert isinstance(exc.value.trigger, FileTrigger), "Trigger is not a
FileTrigger"
+
+ def test_start_trigger_args_are_not_shared_between_tasks(self):
+ """Each task must carry its own trigger arguments.
+
+ ``start_trigger_args`` is a class attribute, so assigning through it
made every task
+ built from this operator advertise the path and timeout of whichever
was constructed
+ last.
+ """
+ with DAG(
+ dag_id="test_start_trigger_args_not_shared",
+ schedule=None,
+ start_date=datetime(2020, 1, 1),
+ ):
+ first = FileSensor(
+ task_id="first", filepath="first.txt", deferrable=True,
start_from_trigger=True, timeout=60
+ )
+ second = FileSensor(
+ task_id="second", filepath="second.txt", deferrable=True,
start_from_trigger=True, timeout=999
+ )
+
+ assert first.start_trigger_args is not second.start_trigger_args
+ assert first.start_trigger_args.trigger_kwargs["filepath"] ==
first.path
+ assert second.start_trigger_args.trigger_kwargs["filepath"] ==
second.path
+ assert first.start_trigger_args.timeout == timedelta(seconds=60)
+ assert second.start_trigger_args.timeout == timedelta(seconds=999)
diff --git a/providers/standard/tests/unit/standard/sensors/test_time.py
b/providers/standard/tests/unit/standard/sensors/test_time.py
index 1989c35d91b..2b44c61d474 100644
--- a/providers/standard/tests/unit/standard/sensors/test_time.py
+++ b/providers/standard/tests/unit/standard/sensors/test_time.py
@@ -135,3 +135,22 @@ class TestTimeSensor:
op.execute_complete(context={}, event={"status": "success"})
except TypeError as e:
pytest.fail(f"TypeError raised: {e}")
+
+ def test_start_trigger_args_are_not_shared_between_tasks(self):
+ """Each task must carry its own trigger arguments.
+
+ ``start_trigger_args`` is a class attribute, so assigning through it
made every task
+ built from this operator advertise the moment of whichever was
constructed last.
+ """
+ with DAG(
+ dag_id="test_start_trigger_args_not_shared",
+ schedule=None,
+ start_date=datetime(2020, 1, 1),
+ ):
+ early = TimeSensor(task_id="early", target_time=time(1, 0),
start_from_trigger=True)
+ late = TimeSensor(task_id="late", target_time=time(23, 0),
start_from_trigger=True)
+
+ assert early.start_trigger_args is not late.start_trigger_args
+ assert early.start_trigger_args.trigger_kwargs["moment"] ==
early.target_datetime
+ assert late.start_trigger_args.trigger_kwargs["moment"] ==
late.target_datetime
+ assert early.target_datetime != late.target_datetime
