.join(TaskInstance.task_instance_note, isouter=True)
        .where(
            TaskInstance.dag_id == dag.dag_id,
            TaskInstance.run_id.in_([dag_run.run_id for dag_run in dag_runs]),
        )
        .group_by(TaskInstance.task_id, TaskInstance.run_id, TaskInstance.state, TaskInstance._try_number)
        .order_by(TaskInstance.task_id, TaskInstance.run_id)
    )
    grouped_tis = {task_id: list(tis) for task_id, tis in itertools.groupby(query, key=lambda ti: ti.task_id)}
    sort_order = conf.get("webserver", "grid_view_sorting_order", fallback="topological")
    if sort_order == "topological":
        sort_children_fn = lambda task_group: task_group.topological_sort()
    elif sort_order == "hierarchical_alphabetical":
        sort_children_fn = lambda task_group: task_group.hierarchical_alphabetical_sort()
    else:
        raise AirflowConfigException(f"Unsupported grid_view_sorting_order: {sort_order}")
    def task_group_to_grid(item, grouped_tis, *, is_parent_mapped: bool):
        if not isinstance(item, TaskGroup):
            def _get_summary(task_instance):
                return {
                    "task_id": task_instance.task_id,
                    "run_id": task_instance.run_id,
                    "state": task_instance.state,
                    "queued_dttm": task_instance.queued_dttm,
                    "start_date": task_instance.start_date,
                    "end_date": task_instance.end_date,
                    "try_number": wwwutils.get_try_count(task_instance._try_number, task_instance.state),
                    "note": task_instance.note,
                }
            def _mapped_summary(ti_summaries):