Fix blank Assets dependency graph from missing Dag nodes

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)

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 @@ def test_data_dependencies_include_team_name_on_task_nodes(self, dag_maker, test
         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