Explicitly cover for shared packages in exception

Author: ephraimbuddy

airflow-core/src/airflow/exceptions.py

@@ -21,13 +21,31 @@
 
 from __future__ import annotations
 
+import os
 from http import HTTPStatus
 from typing import TYPE_CHECKING, NamedTuple
 
 if TYPE_CHECKING:
     from airflow.models import DagRun
 
-from airflow.sdk.exceptions import AirflowException, AirflowNotFoundException
+# When _AIRFLOW__AS_LIBRARY is set, airflow.sdk may not be installed
+# In that case, we define fallback exception classes
+if os.environ.get("_AIRFLOW__AS_LIBRARY"):
+    try:
+        from airflow.sdk.exceptions import AirflowException, AirflowNotFoundException
+    except ImportError:
+        # Fallback exception classes when airflow.sdk is not installed
+        class AirflowException(RuntimeError):
+            """Base exception for Airflow errors."""
+
+            pass
+
+        class AirflowNotFoundException(AirflowException):
+            """Raise when a requested object is not found."""
+
+            pass
+else:
+    from airflow.sdk.exceptions import AirflowException, AirflowNotFoundException
 
 
 class TaskNotFound(AirflowException):

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

@@ -112,7 +112,7 @@ def test_no_target_time(self):
         """Check if BranchDateTimeOperator raises exception on missing target"""
         with pytest.raises(AirflowException):
             BranchDateTimeOperator(
-                task_id="datetime_branch_2",
+                task_id="datetime_branch",
                 follow_task_ids_if_true="branch_1",
                 follow_task_ids_if_false="branch_2",
                 target_upper=None,