feat: Add listener notification in API endpoint when skipped (apache#60585)

Author: kacpermuda

airflow-core/docs/administration-and-deployment/listeners.rst

@@ -96,6 +96,12 @@ of a :class:`~airflow.sdk.execution_time.task_runner.RuntimeTaskInstance` instan
     :start-after: [START howto_listen_ti_failure_task]
     :end-before: [END howto_listen_ti_failure_task]
 
+- ``on_task_instance_skipped``
+
+.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py
+    :language: python
+    :start-after: [START howto_listen_ti_skipped_task]
+    :end-before: [END howto_listen_ti_skipped_task]
 
 Asset Events
 --------------
@@ -199,3 +205,5 @@ List of changes in the listener interfaces since 2.8.0 when they were introduced
 | 3.0.0           | ``on_task_instance_failed``,               | ``session`` argument removed from task instance listeners,                                                                    |
 |                 | ``on_task_instance_success``               | ``task_instance`` object is now an instance of ``RuntimeTaskInstance`` when on worker and ``TaskInstance`` when on API server |
 +-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+
+| 3.2.0           | ``on_task_instance_skipped``               | New listener method added to the interface                                                                                    |
++-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+

airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py

@@ -132,6 +132,8 @@ def _patch_task_instance_state(
                     task_instance=ti,
                     error=f"TaskInstance's state was manually set to `{TaskInstanceState.FAILED}`.",
                 )
+            elif data["new_state"] == TaskInstanceState.SKIPPED:
+                get_listener_manager().hook.on_task_instance_skipped(previous_state=None, task_instance=ti)
         except Exception:
             log.exception("error calling listener")
 

airflow-core/src/airflow/example_dags/plugins/event_listener.py

@@ -30,7 +30,9 @@
 
 # [START howto_listen_ti_running_task]
 @hookimpl
-def on_task_instance_running(previous_state: TaskInstanceState, task_instance: RuntimeTaskInstance):
+def on_task_instance_running(
+    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
+):
     """
     Called when task state changes to RUNNING.
 
@@ -63,7 +65,7 @@ def on_task_instance_running(previous_state: TaskInstanceState, task_instance: R
 # [START howto_listen_ti_success_task]
 @hookimpl
 def on_task_instance_success(
-    previous_state: TaskInstanceState, task_instance: RuntimeTaskInstance | TaskInstance
+    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
 ):
     """
     Called when task state changes to SUCCESS.
@@ -95,7 +97,7 @@ def on_task_instance_success(
 # [START howto_listen_ti_failure_task]
 @hookimpl
 def on_task_instance_failed(
-    previous_state: TaskInstanceState,
+    previous_state: TaskInstanceState | None,
     task_instance: RuntimeTaskInstance | TaskInstance,
     error: None | str | BaseException,
 ):
@@ -133,6 +135,44 @@ def on_task_instance_failed(
 # [END howto_listen_ti_failure_task]
 
 
+# [START howto_listen_ti_skipped_task]
+@hookimpl
+def on_task_instance_skipped(
+    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
+):
+    """
+    Called when a task instance skips itself during execution.
+
+    This hook is called only when a task has started execution and then
+    intentionally skips itself (e.g., by raising AirflowSkipException).
+
+    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
+        - Skips due to trigger rules (e.g., upstream failures)
+        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
+        - Any other situation in which the scheduler decides not to schedule a task for execution
+
+    For comprehensive tracking of skipped tasks, use DAG-level listeners
+    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.
+    """
+    print("Task instance was skipped")
+
+    if isinstance(task_instance, TaskInstance):
+        print("Task instance's state was changed through the API.")
+        return
+
+    context = task_instance.get_template_context()
+    task = context["task"]
+
+    if TYPE_CHECKING:
+        assert task
+
+    print("Task start")
+    print(f"Task:{task}")
+
+
+# [END howto_listen_ti_skipped_task]
+
+
 # [START howto_listen_dagrun_success_task]
 @hookimpl
 def on_dag_run_success(dag_run: DagRun, msg: str):

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -3924,7 +3924,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
         [
             ("success", [TaskInstanceState.SUCCESS]),
             ("failed", [TaskInstanceState.FAILED]),
-            ("skipped", []),
+            ("skipped", [TaskInstanceState.SKIPPED]),
+            ("running", []),
         ],
     )
     def test_patch_task_instance_notifies_listeners(

airflow-core/tests/unit/listeners/class_listener.py

@@ -49,6 +49,10 @@ def on_task_instance_success(self, previous_state, task_instance):
     def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException):
         self.state.append(TaskInstanceState.FAILED)
 
+    @hookimpl
+    def on_task_instance_skipped(self, previous_state, task_instance):
+        self.state.append(TaskInstanceState.SKIPPED)
+
     @hookimpl
     def on_dag_run_running(self, dag_run, msg: str):
         self.state.append(DagRunState.RUNNING)