Move out some exceptions to TaskSDK

airflow-core/newsfragments/54505.significant.rst

@@ -0,0 +1,60 @@
+Move task-level exception imports into the Task SDK
+
+Airflow now sources task-facing exceptions (``AirflowSkipException``, ``TaskDeferred``, etc.) from
+``airflow.sdk.exceptions``. ``airflow.exceptions`` still exposes the same exceptions, but they are
+proxies that emit ``DeprecatedImportWarning`` so Dag authors can migrate before the shim is removed.
+
+**What changed:**
+
+- Runtime code now consistently raises the SDK versions of task-level exceptions.
+- The Task SDK redefines these classes so workers no longer depend on ``airflow-core`` at runtime.
+- ``airflow.providers.common.compat.sdk`` centralizes compatibility imports for providers.
+
+**Behaviour changes:**
+
+- Sensors and other helpers that validate user input now raise ``ValueError`` (instead of
+  ``AirflowException``) when ``poke_interval``/ ``timeout`` arguments are invalid.
+- Importing deprecated exception names from ``airflow.exceptions`` logs a warning directing users to
+  the SDK import path.
+
+**Exceptions now provided by ``airflow.sdk.exceptions``:**
+
+- ``AirflowException`` and ``AirflowNotFoundException``
+- ``AirflowRescheduleException`` and ``AirflowSensorTimeout``
+- ``AirflowSkipException``, ``AirflowFailException``, ``AirflowTaskTimeout``, ``AirflowTaskTerminated``
+- ``TaskDeferred``, ``TaskDeferralTimeout``, ``TaskDeferralError``
+- ``DagRunTriggerException`` and ``DownstreamTasksSkipped``
+- ``AirflowDagCycleException`` and ``AirflowInactiveAssetInInletOrOutletException``
+- ``ParamValidationError``, ``DuplicateTaskIdFound``, ``TaskAlreadyInTaskGroup``, ``TaskNotFound``, ``XComNotFound``
+
+**Backward compatibility:**
+
+- Existing Dags/operators that still import from ``airflow.exceptions`` continue to work, though
+  they log warnings.
+- Providers can rely on ``airflow.providers.common.compat.sdk`` to keep one import path that works
+  across supported Airflow versions.
+
+**Migration:**
+
+- Update custom operators, sensors, and extensions to import exception classes from
+  ``airflow.sdk.exceptions`` (or from the provider compat shim).
+- Adjust custom validation code to expect ``ValueError`` for invalid sensor arguments if it
+  previously caught ``AirflowException``.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [ ] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [x] Code interface changes
+
+* Migration rules needed
+
+  * Import task-level exceptions such as ``AirflowSkipException``, ``TaskDeferred``,
+    ``AirflowFailException``, etc. from ``airflow.sdk.exceptions`` instead of ``airflow.exceptions``.
+  * Update custom sensors/operators that validated arguments by catching ``AirflowException`` to
+    expect ``ValueError`` for invalid ``poke_interval`` / ``timeout`` inputs.

airflow-core/src/airflow/dag_processing/dagbag.py

@@ -41,10 +41,8 @@
     AirflowClusterPolicyError,
     AirflowClusterPolicySkipDag,
     AirflowClusterPolicyViolation,
-    AirflowDagCycleException,
     AirflowDagDuplicatedIdException,
     AirflowException,
-    AirflowTaskTimeout,
     UnknownExecutorException,
 )
 from airflow.executors.executor_loader import ExecutorLoader
@@ -119,6 +117,8 @@ def timeout(seconds=1, error_message="Timeout"):
     def handle_timeout(signum, frame):
         """Log information and raises AirflowTaskTimeout."""
         log.error("Process timed out, PID: %s", str(os.getpid()))
+        from airflow.sdk.exceptions import AirflowTaskTimeout
+
         raise AirflowTaskTimeout(error_message)
 
     try:
@@ -588,6 +588,7 @@ def bag_dag(self, dag: DAG):
         except Exception as e:
             self.log.exception(e)
             raise AirflowClusterPolicyError(e)
+        from airflow.sdk.exceptions import AirflowDagCycleException
 
         try:
             prev_dag = self.dags.get(dag.dag_id)

airflow-core/src/airflow/dag_processing/processor.py

@@ -36,7 +36,7 @@
 )
 from airflow.configuration import conf
 from airflow.dag_processing.dagbag import DagBag
-from airflow.exceptions import TaskNotFound
+from airflow.sdk.exceptions import TaskNotFound
 from airflow.sdk.execution_time.comms import (
     ConnectionResult,
     DeleteVariable,

airflow-core/src/airflow/example_dags/example_skip_dag.py

@@ -24,9 +24,9 @@
 
 import pendulum
 
-from airflow.exceptions import AirflowSkipException
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.sdk import DAG, BaseOperator, TriggerRule
+from airflow.sdk.exceptions import AirflowSkipException
 
 if TYPE_CHECKING:
     from airflow.sdk import Context