Move DagIsPaused exception to standard provider

airflow-core/src/airflow/exceptions.py

@@ -246,17 +246,6 @@ class AirflowTimetableInvalid(AirflowException):
     """Raise when a DAG has an invalid timetable."""
 
 
-class DagIsPaused(AirflowException):
-    """Raise when a dag is paused and something tries to run it."""
-
-    def __init__(self, dag_id: str) -> None:
-        super().__init__(dag_id)
-        self.dag_id = dag_id
-
-    def __str__(self) -> str:
-        return f"Dag {self.dag_id} is paused"
-
-
 class DagNotFound(AirflowNotFoundException):
     """Raise when a DAG is not available in the system."""
 

providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py

@@ -60,14 +60,24 @@
         from airflow.utils.context import Context
 
 if AIRFLOW_V_3_0_PLUS:
-    from airflow.exceptions import DagIsPaused
     from airflow.sdk import BaseOperatorLink
     from airflow.sdk.execution_time.xcom import XCom
 else:
     from airflow.models import XCom  # type: ignore[no-redef]
     from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
 
 
+class DagIsPaused(AirflowException):
+    """Raise when a dag is paused and something tries to run it."""
+
+    def __init__(self, dag_id: str) -> None:
+        super().__init__(dag_id)
+        self.dag_id = dag_id
+
+    def __str__(self) -> str:
+        return f"Dag {self.dag_id} is paused"
+
+
 class TriggerDagRunLink(BaseOperatorLink):
     """
     Operator link for TriggerDagRunOperator.

providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py

@@ -30,7 +30,7 @@
 from airflow.models.dagrun import DagRun
 from airflow.models.log import Log
 from airflow.models.taskinstance import TaskInstance
-from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.providers.standard.operators.trigger_dagrun import DagIsPaused, TriggerDagRunOperator
 from airflow.providers.standard.triggers.external_task import DagStateTrigger
 from airflow.utils import timezone
 from airflow.utils.session import create_session
@@ -41,7 +41,7 @@
 from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
 
 if AIRFLOW_V_3_0_PLUS:
-    from airflow.exceptions import DagIsPaused, DagRunTriggerException
+    from airflow.exceptions import DagRunTriggerException
 
 pytestmark = pytest.mark.db_test