Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py (Python)

Code before:
from airflow.api_fastapi.core_api.services.ui.task_group import get_task_group_children_getter
from airflow.models.taskmap import TaskMap
from airflow.serialization.definitions.baseoperator import SerializedBaseOperator
from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator
from airflow.serialization.definitions.taskgroup import SerializedTaskGroup

log = structlog.get_logger(logger_name=__name__)


def _merge_node_dicts(current: list[dict[str, any]], new: list[dict[str, any]] | None) -> None:
    """Merge node dictionaries from different DAG versions, handling structure changes."""
    # Handle None case - can occur when merging old DAG versions
    # where a task was converted to a TaskGroup or vice versa
    if new is None:
        return

    current_ids = {node["id"] for node in current}
    for node in new:
        if node["id"] in current_ids:
            current_node = _get_node_by_id(current, node["id"])
            # Only merge children if both nodes have children to avoid TypeError
            # This handles cases where a task is converted to a TaskGroup between versions
            current_children = current_node.get("children")
            node_children = node.get("children")
            if current_children is not None and node_children is not None:
                _merge_node_dicts(current_children, node_children)
        else:
            current.append(node)


def _get_node_by_id(nodes: list[dict[str, any]], node_id: str) -> dict[str, any]:
    """Find a node by its ID in a list of nodes."""
    for node in nodes:
        if node["id"] == node_id:
            return node
    return {}


def agg_state(states):
    states = Counter(states)
    for state in state_priority:
        if state in states:
            return state
    return None


def _get_aggs_for_node(detail):
    states = [x["state"] for x in detail]
    try:
        min_start_date = min(x["start_date"] for x in detail if x["start_date"])
    except ValueError:

Review comment (suggestion):
I don't think this logic is correct.

What if `current_children` is None, but `node_children` is not None ? This won't do anything and be skipped, while we actually expect the children from the 'node_children' to be merged into the 'current_node'. (an array should be created and children should be appended to that).

Instead I think we should skip if both are None, but all other cases, merge and provider an empty list default to None arrays.

Code after:
from airflow.api_fastapi.common.parameters import state_priority
from airflow.api_fastapi.core_api.services.ui.task_group import get_task_group_children_getter
from airflow.models.taskmap import TaskMap
from airflow.serialization.definitions.baseoperator import SerializedBaseOperator
from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator
from airflow.serialization.definitions.taskgroup import SerializedTaskGroup

log = structlog.get_logger(logger_name=__name__)


def _merge_node_dicts(current: list[dict[str, Any]], new: list[dict[str, Any]] | None) -> None:
    """Merge node dictionaries from different DAG versions, handling structure changes."""
    # Handle None case - can occur when merging old DAG versions
    # where a TaskGroup was converted to a task or vice versa
    if new is None:
        return

    current_nodes_by_id = {node["id"]: node for node in current}
    for node in new:
        node_id = node["id"]
        current_node = current_nodes_by_id.get(node_id)
        if current_node is not None:
            # Only merge children if current node already has children
            # This preserves the structure of the latest DAG version
            if current_node.get("children") is not None:
                _merge_node_dicts(current_node["children"], node.get("children"))
        else:
            current.append(node)
            current_nodes_by_id[node_id] = node


def agg_state(states):
    states = Counter(states)
    for state in state_priority:
        if state in states:
            return state
    return None


def _get_aggs_for_node(detail):
    states = [x["state"] for x in detail]
    try:
        min_start_date = min(x["start_date"] for x in detail if x["start_date"])
    except ValueError:
        min_start_date = None
    try:
        max_end_date = max(x["end_date"] for x in detail if x["end_date"])
    except ValueError:
        max_end_date = None
    return {
        "state": agg_state(states),