https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""Mark DagRun failed."""
        dag_id = request.form.get("dag_id")
        dag_run_id = request.form.get("dag_run_id")
        confirmed = request.form.get("confirmed") == "true"
        return self._mark_dagrun_state_as_failed(dag_id, dag_run_id, confirmed)
    @expose("/dagrun_success", methods=["POST"])
    @auth.has_access_dag("PUT", DagAccessEntity.RUN)
    @action_logging
    def dagrun_success(self):
        """Mark DagRun success."""
        dag_id = request.form.get("dag_id")
        dag_run_id = request.form.get("dag_run_id")
        confirmed = request.form.get("confirmed") == "true"
        return self._mark_dagrun_state_as_success(dag_id, dag_run_id, confirmed)
    @expose("/dagrun_queued", methods=["POST"])
    @auth.has_access_dag("PUT", DagAccessEntity.RUN)
    @action_logging
    def dagrun_queued(self):
        """Queue DagRun so tasks that haven't run yet can be started."""
        dag_id = request.form.get("dag_id")
        dag_run_id = request.form.get("dag_run_id")
        confirmed = request.form.get("confirmed") == "true"
        return self._mark_dagrun_state_as_queued(dag_id, dag_run_id, confirmed)
    @expose("/dagrun_details")
    def dagrun_details(self):
        """Redirect to the GRID DAGRun page. This is avoids breaking links."""
        dag_id = request.args.get("dag_id")
        run_id = request.args.get("run_id")
        return redirect(url_for("Airflow.grid", dag_id=dag_id, dag_run_id=run_id))
    def _mark_task_instance_state(