Better typing for BaseOperator defer (#39742)

This adds typing for the `defer` method, and covers the core deferrable
sensors. It also fixes 1 error in a databricks provider, but
leaves the rest of the provider ecosystem alone.

(cherry picked from commit f18e634)

airflow/models/baseoperator.py

@@ -41,6 +41,7 @@
     Callable,
     Collection,
     Iterable,
+    NoReturn,
     Sequence,
     TypeVar,
     Union,
@@ -1674,7 +1675,7 @@ def defer(
         method_name: str,
         kwargs: dict[str, Any] | None = None,
         timeout: timedelta | None = None,
-    ):
+    ) -> NoReturn:
         """
         Mark this Operator "deferred", suspending its execution until the provided trigger fires an event.
 

airflow/sensors/date_time.py

@@ -18,7 +18,7 @@
 from __future__ import annotations
 
 import datetime
-from typing import TYPE_CHECKING, Sequence
+from typing import TYPE_CHECKING, NoReturn, Sequence
 
 from airflow.sensors.base import BaseSensorOperator
 from airflow.triggers.temporal import DateTimeTrigger
@@ -90,13 +90,13 @@ class DateTimeSensorAsync(DateTimeSensor):
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
 
-    def execute(self, context: Context):
+    def execute(self, context: Context) -> NoReturn:
         trigger = DateTimeTrigger(moment=timezone.parse(self.target_time))
         self.defer(
             trigger=trigger,
             method_name="execute_complete",
         )
 
-    def execute_complete(self, context, event=None):
+    def execute_complete(self, context, event=None) -> None:
         """Execute when the trigger fires - returns immediately."""
         return None

airflow/sensors/time_delta.py

@@ -17,7 +17,7 @@
 # under the License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NoReturn
 
 from airflow.exceptions import AirflowSkipException
 from airflow.sensors.base import BaseSensorOperator
@@ -66,7 +66,7 @@ class TimeDeltaSensorAsync(TimeDeltaSensor):
 
     """
 
-    def execute(self, context: Context):
+    def execute(self, context: Context) -> NoReturn:
         target_dttm = context["data_interval_end"]
         target_dttm += self.delta
         try:
@@ -78,6 +78,6 @@ def execute(self, context: Context):
 
         self.defer(trigger=trigger, method_name="execute_complete")
 
-    def execute_complete(self, context, event=None):
+    def execute_complete(self, context, event=None) -> None:
         """Execute for when the trigger fires - return immediately."""
         return None

airflow/sensors/time_sensor.py

@@ -18,7 +18,7 @@
 from __future__ import annotations
 
 import datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NoReturn
 
 from airflow.sensors.base import BaseSensorOperator
 from airflow.triggers.temporal import DateTimeTrigger
@@ -72,7 +72,7 @@ def __init__(self, *, target_time: datetime.time, **kwargs) -> None:
 
         self.target_datetime = timezone.convert_to_utc(aware_time)
 
-    def execute(self, context: Context):
+    def execute(self, context: Context) -> NoReturn:
         trigger = DateTimeTrigger(moment=self.target_datetime)
         self.defer(
             trigger=trigger,