fixup! fixup! Move out some exceptions to TaskSDK

Author: ephraimbuddy

airflow-core/src/airflow/exceptions.py

@@ -24,24 +24,12 @@
 from http import HTTPStatus
 from typing import TYPE_CHECKING, NamedTuple
 
+from airflow.sdk.exceptions import AirflowException, AirflowNotFoundException
+
 if TYPE_CHECKING:
     from airflow.models import DagRun
 
 
-class AirflowException(Exception):
-    """
-    Base class for all Airflow's errors.
-
-    Each custom exception should be derived from this class.
-    """
-
-    status_code = HTTPStatus.INTERNAL_SERVER_ERROR
-
-    def serialize(self):
-        cls = self.__class__
-        return f"{cls.__module__}.{cls.__name__}", (str(self),), {}
-
-
 class AirflowBadRequest(AirflowException):
     """Raise when the application or server cannot handle the request."""
 
@@ -52,12 +40,6 @@ class TaskNotFound(AirflowException):
     """Raise when a Task is not available in the system."""
 
 
-class AirflowNotFoundException(AirflowException):
-    """Raise when the requested object/resource is not available in the system."""
-
-    status_code = HTTPStatus.NOT_FOUND
-
-
 class AirflowConfigException(AirflowException):
     """Raise when there is configuration problem."""
 
@@ -66,10 +48,6 @@ class InvalidStatsNameException(AirflowException):
     """Raise when name of the stats is invalid."""
 
 
-class AirflowFailException(AirflowException):
-    """Raise when the task should be failed without retrying."""
-
-
 class AirflowOptionalProviderFeatureException(AirflowException):
     """Raise by providers when imports are missing for optional provider features."""
 

airflow-core/tests/unit/models/test_taskinstance.py

@@ -35,12 +35,7 @@
 
 from airflow import settings
 from airflow._shared.timezones import timezone
-from airflow.exceptions import (
-    AirflowException,
-    AirflowFailException,
-    AirflowInactiveAssetInInletOrOutletException,
-    AirflowSkipException,
-)
+from airflow.exceptions import AirflowException
 from airflow.models.asset import AssetActive, AssetAliasModel, AssetEvent, AssetModel
 from airflow.models.baseoperator import BaseOperator
 from airflow.models.connection import Connection
@@ -69,6 +64,11 @@
 from airflow.sdk.definitions.asset import Asset, AssetAlias
 from airflow.sdk.definitions.param import process_params
 from airflow.sdk.definitions.taskgroup import TaskGroup
+from airflow.sdk.exceptions import (
+    AirflowFailException,
+    AirflowInactiveAssetInInletOrOutletException,
+    AirflowSkipException,
+)
 from airflow.sdk.execution_time.comms import (
     AssetEventsResult,
 )