pierrejeambrun commented on code in PR #69833:
URL: https://github.com/apache/airflow/pull/69833#discussion_r3614419066
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/structure.py:
##########
@@ -104,82 +96,4 @@ def structure_data(
"edges": edges,
}
- if external_dependencies:
- entry_node_ref = nodes[0] if nodes else None
- exit_node_ref = nodes[-1] if nodes else None
-
- start_edges: list[dict] = []
- end_edges: list[dict] = []
-
- readable_dag_ids = readable_dags_filter.value
- for dependency_dag_id, dependencies in
sorted(SerializedDagModel.get_dag_dependencies().items()):
- if readable_dag_ids is not None and dependency_dag_id not in
readable_dag_ids:
- continue
- for dependency in dependencies:
- # Dependencies not related to `dag_id` are ignored
- if dependency_dag_id != dag_id and dependency.target != dag_id:
- continue
- # When target is a real Dag ID (not a type label), hide it
- # if the caller cannot read that Dag.
- if (
- readable_dag_ids is not None
- and dependency.target != dependency.dependency_type
- and dependency.target not in readable_dag_ids
- ):
- continue
-
- # upstream assets are handled by the `get_upstream_assets`
function.
- if dependency.target != dependency.dependency_type and
dependency.dependency_type in [
- "asset-alias",
- "asset",
- ]:
- continue
-
- # Add edges
- # start dependency
- if (
- dependency.source == dependency.dependency_type or
dependency.target == dag_id
- ) and entry_node_ref:
- start_edges.append({"source_id": dependency.node_id,
"target_id": entry_node_ref["id"]})
-
- # end dependency
- elif (
- dependency.target == dependency.dependency_type or
dependency.source == dag_id
- ) and exit_node_ref:
- end_edges.append(
- {
- "source_id": exit_node_ref["id"],
- "target_id": dependency.node_id,
- "resolved_from_alias":
dependency.source.replace("asset-alias:", "", 1)
- if dependency.source.startswith("asset-alias:")
- else None,
- }
- )
-
- # Add nodes
- nodes.append(
- {
- "id": dependency.node_id,
- "label": dependency.label,
- "type": dependency.dependency_type,
- }
- )
-
- if (asset_expression := serialized_dag.dag_model.asset_expression) and
entry_node_ref:
- try:
- upstream_asset_nodes, upstream_asset_edges =
get_upstream_assets(
- asset_expression, entry_node_ref["id"]
- )
- except TypeError as e:
- raise HTTPException(
- status.HTTP_400_BAD_REQUEST,
- f"Malformed asset_expression in Dag {dag_id!r} version
{version_number}: {e}",
- ) from e
- data["nodes"] += upstream_asset_nodes
- data["edges"] += upstream_asset_edges
-
- data["edges"] += start_edges + end_edges
-
- bind_output_assets_to_tasks(data["edges"], serialized_dag, version_number,
session)
-
return StructureDataResponse(**data)
Review Comment:
With `bind_output_assets_to_tasks` gone, produced assets are no longer bound
to the specific producing task — in "show all" the merged graph attaches them
at the `dag:` boundary via `get_scheduling_dependencies`. Intended UX
simplification, or a regression from the per-task binding?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
