This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 87bf0236d92 Fix blank Assets dependency graph from missing Dag nodes
(#69171)
87bf0236d92 is described below
commit 87bf0236d924c9dd9feb19e21d8a236844905848
Author: davidnzhang <[email protected]>
AuthorDate: Fri Jul 3 01:53:31 2026 +1000
Fix blank Assets dependency graph from missing Dag nodes (#69171)
---
.../core_api/services/ui/dependencies.py | 13 ++++++++++
.../core_api/routes/ui/test_dependencies.py | 30 ++++++++++++++++++++++
2 files changed, 43 insertions(+)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
index eb4e8e20910..20ffc78008a 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
@@ -135,6 +135,19 @@ def get_scheduling_dependencies(readable_dag_ids: set[str]
| None = None) -> dic
target = dep.target if ":" in dep.target else
f"dag:{dep.target}"
edge_tuples.add((source, target))
+ # Create missing ``dag:`` nodes which may have been skipped by the loop
above.
+ # A DAG referenced only as a trigger target or a sensor source may have no
+ # scheduling dependencies of its own. Without this loop, these DAGs will
not be
+ # materialised and will result in dangling edges.
+ for source, target in edge_tuples:
+ for endpoint in (source, target):
+ if endpoint.startswith("dag:") and endpoint not in nodes_dict:
+ nodes_dict[endpoint] = {
+ "id": endpoint,
+ "label": endpoint.removeprefix("dag:"),
+ "type": "dag",
+ }
+
dag_ids = [node["label"] for node in nodes_dict.values() if node["type"]
== "dag"]
if dag_ids:
dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping(dag_ids)
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
index 513290accb9..eca2fa00875 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
@@ -147,6 +147,12 @@ def expected_primary_component_response(asset1_id):
"type": "dag",
"team": None,
},
+ {
+ "id": "dag:other_dag",
+ "label": "other_dag",
+ "type": "dag",
+ "team": None,
+ },
],
}
@@ -551,3 +557,27 @@ class TestGetDependencies:
task_nodes = [node for node in result["nodes"] if node["type"] ==
"task"]
assert len(task_nodes) == 1
assert task_nodes[0]["team"] == "data-team"
+
+ def
test_trigger_target_without_scheduling_dependencies_is_materialised(self,
dag_maker, test_client):
+ """A DAG triggered via TriggerDagRunOperator that has no scheduling
dependencies must still appear as a node"""
+ with dag_maker(dag_id="parent_dag", serialized=True):
+ TriggerDagRunOperator(task_id="trigger_child_dag",
trigger_dag_id="child_dag")
+
+ with dag_maker(dag_id="child_dag", serialized=True):
+ EmptyOperator(task_id="some_task")
+
+ dag_maker.sync_dagbag_to_db()
+
+ response = test_client.get("/dependencies")
+ assert response.status_code == 200
+
+ result = response.json()
+ node_ids = {node["id"] for node in result["nodes"]}
+ assert "dag:child_dag" in node_ids
+
+ expected_edge = (
+ "trigger:parent_dag:child_dag:trigger_child_dag",
+ "dag:child_dag",
+ )
+ actual_edges = {(edge["source_id"], edge["target_id"]) for edge in
result["edges"]}
+ assert expected_edge in actual_edges
