diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py index e4f9b02f7d1b0..151aa0ea1461a 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py @@ -104,7 +104,7 @@ def _get_serdag( if dag_version_id is not None: serdag = dag_bag.get_dag(dag_version_id, session=session) if serdag is None: - log.error("No serialized dag found", dag_id=dag_id, version_id=dag_version_id) + log.warning("No serialized dag found", dag_id=dag_id, version_id=dag_version_id) return serdag # Fallback: pre-3.0 upgrade — pick the oldest DagVersion for this dag_id. @@ -115,7 +115,7 @@ def _get_serdag( return None serdag = dag_bag.get_dag(oldest_version_id, session=session) if serdag is None: - log.error("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id) + log.warning("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id) return serdag @@ -429,21 +429,22 @@ def _build_ti_summaries( return None serdag = _get_serdag(dag_bag, dag_id, dag_version_id, session) - if TYPE_CHECKING: - assert serdag def get_node_summaries() -> Iterable[dict[str, Any]]: yielded_task_ids: set[str] = set() - for node, _ in _find_aggregates( - node=serdag.task_group, - parent_node=None, - ti_details=ti_details, - ): - if node["type"] in {"task", "mapped_task"}: - yielded_task_ids.add(node["task_id"]) - if node["type"] == "task": - node["child_states"] = None - yield node + # serdag is None once `airflow db clean` removes the version's serialized Dag row. Leaving + # yielded_task_ids empty then routes every task through the flat-node path below. + if serdag is not None: + for node, _ in _find_aggregates( + node=serdag.task_group, + parent_node=None, + ti_details=ti_details, + ): + if node["type"] in {"task", "mapped_task"}: + yielded_task_ids.add(node["task_id"]) + if node["type"] == "task": + node["child_states"] = None + yield node missing_task_ids = set(ti_details.keys()) - yielded_task_ids for task_id in sorted(missing_task_ids): detail = ti_details[task_id] diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py index ad4450903df87..b67a4c7d8b789 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py @@ -23,7 +23,7 @@ import pendulum import pytest -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.orm import Session from airflow._shared.timezones import timezone @@ -31,6 +31,7 @@ from airflow.models.dag_version import DagVersion from airflow.models.dagbag import DBDagBag from airflow.models.dagrun import DagRun, DagRunNote +from airflow.models.serialized_dag import SerializedDagModel from airflow.models.taskinstance import TaskInstance, TaskInstanceNote from airflow.providers.standard.operators.empty import EmptyOperator from airflow.providers.standard.operators.python import PythonOperator @@ -1260,6 +1261,55 @@ def test_grid_ti_summaries_stream_skips_missing_runs(self, session, test_client) assert len(summaries) == 1 assert summaries[0]["run_id"] == "run_1" + def test_grid_ti_summaries_stream_continues_past_cleaned_dag_version(self, session, test_client): + """A run whose version was cleaned still streams, and does not cut the runs after it.""" + cleaned_version = session.scalar( + select(DagVersion).where(DagVersion.dag_id == DAG_ID_5, DagVersion.version_number == 1) + ) + session.execute( + delete(SerializedDagModel).where(SerializedDagModel.dag_version_id == cleaned_version.id) + ) + session.commit() + + response = test_client.get( + f"/grid/ti_summaries/{DAG_ID_5}", params={"run_ids": ["run_5_1", "run_5_2"]} + ) + assert response.status_code == 200 + summaries = {summary["run_id"]: summary for summary in self._parse_ndjson(response)} + assert set(summaries) == {"run_5_1", "run_5_2"} + assert {node["task_id"] for node in summaries["run_5_1"]["task_instances"]} == { + "task_a", + "task_b", + "task_c", + "task_d", + "task_f", + } + # The run on the surviving version is unaffected. + assert {node["task_id"] for node in summaries["run_5_2"]["task_instances"]} == { + "task_a", + "task_b", + "task_c", + "task_d", + "task_e", + } + + def test_grid_ti_summaries_stream_without_serialized_dag_flattens_groups(self, session, test_client): + """With no structure to walk, a grouped Dag's tasks come back as flat leaf nodes.""" + session.execute(delete(SerializedDagModel).where(SerializedDagModel.dag_id == DAG_ID)) + session.commit() + + response = test_client.get(f"/grid/ti_summaries/{DAG_ID}", params={"run_ids": ["run_1"]}) + assert response.status_code == 200 + [summary] = self._parse_ndjson(response) + assert {node["task_id"]: node["state"] for node in summary["task_instances"]} == { + TASK_ID: "success", + "mapped_task_group.subtask": "success", + f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}": "success", + f"{TASK_GROUP_ID}.{INNER_TASK_GROUP}.{INNER_TASK_GROUP_SUB_TASK}": "success", + MAPPED_TASK_ID_2: "success", + } + assert all(node["child_states"] is None for node in summary["task_instances"]) + def test_grid_ti_summaries_stream_empty_run_ids(self, session, test_client): """Streaming endpoint with no run_ids returns an empty body.""" session.commit()