qry = session.execute(
            select(
                final_task_instance_query_result.c.dag_id,
                final_task_instance_query_result.c.state,
                final_task_instance_query_result.c.is_dag_running,
                sqla.func.count(),
            )
            .group_by(
                final_task_instance_query_result.c.dag_id,
                final_task_instance_query_result.c.state,
                final_task_instance_query_result.c.is_dag_running,
            )
            .order_by(
                final_task_instance_query_result.c.dag_id,
                final_task_instance_query_result.c.is_dag_running.desc(),
            )
        )
        data = get_task_stats_from_query(qry)
        payload: dict[str, list[dict[str, Any]]] = collections.defaultdict(list)
        for dag_id in filter_dag_ids:
            for state in State.task_states:
                count = data.get(dag_id, {}).get(state, 0)
                payload[dag_id].append({"state": state, "count": count})
        return flask.json.jsonify(payload)
    @expose("/last_dagruns", methods=["POST"])
    @auth.has_access(
        [
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
        ]