fix(sensors): move core async sensor trigger initialization to __init…

…__ if possible

airflow/sensors/date_time.py

@@ -85,9 +85,13 @@ class DateTimeSensorAsync(DateTimeSensor):
     :param target_time: datetime after which the job succeeds. (templated)
     """
 
+    def __init__(self, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.trigger = DateTimeTrigger(moment=timezone.parse(self.target_time))
+
     def execute(self, context: Context):
         self.defer(
-            trigger=DateTimeTrigger(moment=timezone.parse(self.target_time)),
+            trigger=self.trigger,
             method_name="execute_complete",
         )
 

airflow/sensors/time_delta.py

@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+from airflow.exceptions import AirflowSkipException
 from airflow.sensors.base import BaseSensorOperator
 from airflow.triggers.temporal import DateTimeTrigger
 from airflow.utils import timezone
@@ -64,7 +65,14 @@ class TimeDeltaSensorAsync(TimeDeltaSensor):
     def execute(self, context: Context):
         target_dttm = context["data_interval_end"]
         target_dttm += self.delta
-        self.defer(trigger=DateTimeTrigger(moment=target_dttm), method_name="execute_complete")
+        try:
+            trigger = DateTimeTrigger(moment=target_dttm)
+        except (TypeError, ValueError) as e:
+            if self.soft_fail:
+                raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e
+            raise
+
+        self.defer(trigger=trigger, method_name="execute_complete")
 
     def execute_complete(self, context, event=None):
         """Execute for when the trigger fires - return immediately."""

airflow/sensors/time_sensor.py

@@ -68,10 +68,11 @@ def __init__(self, *, target_time, **kwargs):
         )
 
         self.target_datetime = timezone.convert_to_utc(aware_time)
+        self.trigger = DateTimeTrigger(moment=self.target_datetime)
 
     def execute(self, context: Context):
         self.defer(
-            trigger=DateTimeTrigger(moment=self.target_datetime),
+            trigger=self.trigger,
             method_name="execute_complete",
         )