Attempt2: Add sensor deprecation

airflow-core/src/airflow/sensors/__init__.py

@@ -26,8 +26,12 @@
 
 from airflow.utils.deprecation_tools import add_deprecated_classes
 
-# TODO: Add definition from Task SDK here and remove `base.py` file
 __deprecated_classes = {
+    "base": {
+        "BaseSensorOperator": "airflow.sdk.bases.sensor.BaseSensorOperator",
+        "PokeReturnValue": "airflow.sdk.bases.sensor.PokeReturnValue",
+        "poke_mode_only": "airflow.sdk.bases.sensor.poke_mode_only",
+    },
     "python":{
         "PythonSensor": "airflow.providers.standard.sensors.python.PythonSensor",
     },

providers/standard/src/airflow/providers/standard/sensors/date_time.py

@@ -20,7 +20,7 @@
 import datetime
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, NoReturn
+from typing import TYPE_CHECKING, Any
 
 from airflow.providers.standard.triggers.temporal import DateTimeTrigger
 from airflow.providers.standard.version_compat import AIRFLOW_V_3_0_PLUS
@@ -147,7 +147,7 @@ def __init__(
                 end_from_trigger=self.end_from_trigger,
             )
 
-    def execute(self, context: Context) -> NoReturn:
+    def execute(self, context: Context) -> None:
         self.defer(
             method_name="execute_complete",
             trigger=DateTimeTrigger(

providers/standard/src/airflow/providers/standard/sensors/external_task.py

@@ -46,7 +46,7 @@
 if AIRFLOW_V_3_0_PLUS:
     from airflow.sdk.bases.sensor import BaseSensorOperator
 else:
-    from airflow.sensors.base import BaseSensorOperator
+    from airflow.sensors.base import BaseSensorOperator  # type:ignore[no-redef]
     from airflow.utils.session import NEW_SESSION, provide_session
 
 if TYPE_CHECKING:

providers/standard/src/airflow/providers/standard/sensors/time_delta.py

@@ -20,7 +20,7 @@
 import warnings
 from datetime import datetime, timedelta
 from time import sleep
-from typing import TYPE_CHECKING, Any, NoReturn
+from typing import TYPE_CHECKING, Any
 
 from deprecated.classic import deprecated
 from packaging.version import Version
@@ -106,7 +106,7 @@ def poke(self, context: Context) -> bool:
     Asynchronous execution
     """
 
-    def execute(self, context: Context) -> bool | NoReturn:
+    def execute(self, context: Context) -> bool | None:
         """
         Depending on the deferrable flag, either execute the sensor in a blocking way or defer it.
 
@@ -147,6 +147,8 @@ def execute(self, context: Context) -> bool | NoReturn:
             timeout=timeout,
         )
 
+        return None
+
     def execute_complete(self, context: Context, event: Any = None) -> None:
         """Handle the event when the trigger fires and return immediately."""
         return None