Reimplement Sentry integration in Task SDK

airflow-core/docs/extra-packages-ref.rst

@@ -426,10 +426,12 @@ Group extras
 The group extras are convenience extras. Such extra installs many optional dependencies together.
 It is not recommended to use it in production, but it is useful for CI, development and testing purposes.
 
-+-----------+------------------------------------------+---------------------------------------------------+
-| extra     | install command                          | enables                                           |
-+===========+==========================================+===================================================+
-| all       | ``pip install apache-airflow[all]``      | All optional dependencies including all providers |
-+-----------+------------------------------------------+---------------------------------------------------+
-| all-core  | ``pip install apache-airflow[all-core]`` | All optional core dependencies                    |
-+-----------+------------------------------------------+---------------------------------------------------+
++--------------+----------------------------------------------+---------------------------------------------------+
+| extra        | install command                              | enables                                           |
++==============+==============================================+===================================================+
+| all          | ``pip install apache-airflow[all]``          | All optional dependencies including all providers |
++--------------+----------------------------------------------+---------------------------------------------------+
+| all-core     | ``pip install apache-airflow[all-core]``     | All optional core dependencies                    |
++--------------+----------------------------------------------+---------------------------------------------------+
+| all-task-sdk | ``pip install apache-airflow[all-task-sdk]`` | All optional task SDK dependencies                |
++--------------+----------------------------------------------+---------------------------------------------------+

airflow-core/pyproject.toml

@@ -178,16 +178,11 @@ dependencies = [
 "otel" = [
     "opentelemetry-exporter-prometheus>=0.47b0",
 ]
-"sentry" = [
-    "blinker>=1.1",
-    # Apparently sentry needs flask to be installed to work properly
-    "sentry-sdk[flask]>=2.30.0",
-]
 "statsd" = [
     "statsd>=3.3.0",
 ]
 "all" = [
-    "apache-airflow-core[graphviz,kerberos,otel,sentry,statsd]"
+    "apache-airflow-core[graphviz,kerberos,otel,statsd]"
 ]
 
 [project.scripts]

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import uuid
+from collections.abc import Iterable
 from datetime import timedelta
 from enum import Enum
 from typing import Annotated, Any, Literal
@@ -355,6 +356,12 @@ class TaskStatesResponse(BaseModel):
     task_states: dict[str, Any]
 
 
+class TaskBreadcrumbsResponse(BaseModel):
+    """Response for task breadcrumbs."""
+
+    breadcrumbs: Iterable[dict[str, Any]]
+
+
 class InactiveAssetsResponse(BaseModel):
     """Response for inactive assets."""
 

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -43,6 +43,7 @@
 from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
     InactiveAssetsResponse,
     PrevSuccessfulDagRunResponse,
+    TaskBreadcrumbsResponse,
     TaskStatesResponse,
     TIDeferredStatePayload,
     TIEnterRunningPayload,
@@ -68,7 +69,7 @@
 from airflow.sdk.definitions.asset import Asset, AssetUniqueKey
 from airflow.serialization.serialized_objects import SerializedDAG
 from airflow.task.trigger_rule import TriggerRule
-from airflow.utils.state import DagRunState, TaskInstanceState
+from airflow.utils.state import DagRunState, TaskInstanceState, TerminalTIState
 
 if TYPE_CHECKING:
     from sqlalchemy.sql.dml import Update
@@ -917,6 +918,21 @@ def get_task_instance_states(
     return TaskStatesResponse(task_states=run_id_task_state_map)
 
 
+@router.get("/breadcrumbs", status_code=status.HTTP_200_OK)
+def get_task_instance_breadcrumbs(dag_id: str, run_id: str, session: SessionDep) -> TaskBreadcrumbsResponse:
+    result = session.execute(
+        select(TI.task_id, TI.map_index, TI.state, TI.operator, TI.duration)
+        .where(TI.dag_id == dag_id, TI.run_id == run_id, TI.state.in_(TerminalTIState))
+        .order_by(TI.task_id, TI.map_index)
+    ).mappings()
+
+    def _iter_breadcrumbs() -> Iterator[dict[str, Any]]:
+        for row in result:
+            yield {str(k): v for k, v in row.items()}
+
+    return TaskBreadcrumbsResponse(breadcrumbs=_iter_breadcrumbs())
+
+
 def _is_eligible_to_retry(state: str, try_number: int, max_tries: int) -> bool:
     """Is task instance is eligible for retry."""
     if state == TaskInstanceState.RESTARTING:

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -2518,6 +2518,75 @@ def add_one(x):
         assert response.json() == {"task_states": {dr.run_id: expected}}
 
 
+class TestGetTaskInstanceBreadcrumbs:
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    @pytest.fixture(autouse=True)
+    def dag_run(self, dag_maker, session):
+        with dag_maker(session=session):
+            for name in TaskInstanceState._member_names_:
+                EmptyOperator(task_id=name)
+        return dag_maker.create_dagrun(state="running")
+
+    @pytest.fixture(autouse=True)
+    def task_instances(self, dag_run, session):
+        tis = {ti.task_id: ti for ti in dag_run.task_instances}
+        for name, value in TaskInstanceState._member_map_.items():
+            tis[name].state = value
+        session.commit()
+        return tis
+
+    def test_get_breadcrumbs(self, client, dag_run):
+        response = client.get(
+            "/execution/task-instances/breadcrumbs",
+            params={"dag_id": dag_run.dag_id, "run_id": dag_run.run_id},
+        )
+        assert response.status_code == 200
+        assert response.json() == {  # Should find tis with terminal states.
+            "breadcrumbs": [
+                {
+                    "duration": None,
+                    "map_index": -1,
+                    "operator": "EmptyOperator",
+                    "state": "failed",
+                    "task_id": "FAILED",
+                },
+                {
+                    "duration": None,
+                    "map_index": -1,
+                    "operator": "EmptyOperator",
+                    "state": "removed",
+                    "task_id": "REMOVED",
+                },
+                {
+                    "duration": None,
+                    "map_index": -1,
+                    "operator": "EmptyOperator",
+                    "state": "skipped",
+                    "task_id": "SKIPPED",
+                },
+                {
+                    "duration": None,
+                    "map_index": -1,
+                    "operator": "EmptyOperator",
+                    "state": "success",
+                    "task_id": "SUCCESS",
+                },
+                {
+                    "duration": None,
+                    "map_index": -1,
+                    "operator": "EmptyOperator",
+                    "state": "upstream_failed",
+                    "task_id": "UPSTREAM_FAILED",
+                },
+            ]
+        }
+
+
 class TestInvactiveInletsAndOutlets:
     @pytest.mark.parametrize(
         "logical_date",