https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
.join(TaskInstance.dag_run)
                .where(
                    DagRun.start_date >= start_date,
                    func.coalesce(DagRun.end_date, datetime.datetime.utcnow()) <= end_date,
                )
                .group_by(TaskInstance.state)
            ).all()
            data = {
                "dag_run_types": {
                    **{dag_run_type.value: 0 for dag_run_type in DagRunType},
                    **dict(dag_run_types),
                },
                "dag_run_states": {
                    **{dag_run_state.value: 0 for dag_run_state in DagRunState},
                    **dict(dag_run_states),
                },
                "task_instance_states": {
                    "no_status": 0,
                    **{ti_state.value: 0 for ti_state in TaskInstanceState},
                    **{ti_state or "no_status": sum_value for ti_state, sum_value in task_instance_states},
                },
            }
        return (
            htmlsafe_json_dumps(data, separators=(",", ":"), dumps=flask.json.dumps),
            {"Content-Type": "application/json; charset=utf-8"},
        )
    @expose("/object/next_run_datasets/<string:dag_id>")
    @auth.has_access_dag("GET", DagAccessEntity.RUN)
    @auth.has_access_dataset("GET")
    def next_run_datasets(self, dag_id):