Remove legacy dag details page and redirect to grid

airflow/www/views.py

@@ -1316,80 +1316,9 @@ def legacy_dag_details(self):
         return redirect(url_for("Airflow.dag_details", **sanitize_args(request.args)))
 
     @expose("/dags/<string:dag_id>/details")
-    @auth.has_access_dag("GET", DagAccessEntity.RUN)
-    @provide_session
-    def dag_details(self, dag_id, session: Session = NEW_SESSION):
+    def dag_details(self, dag_id):
         """Get Dag details."""
-        from airflow.models.dag import DagOwnerAttributes
-
-        dag = get_airflow_app().dag_bag.get_dag(dag_id, session=session)
-        dag_model = DagModel.get_dagmodel(dag_id, session=session)
-        if not dag:
-            flash(f'DAG "{dag_id}" seems to be missing.', "error")
-            return redirect(url_for("Airflow.index"))
-
-        wwwutils.check_import_errors(dag.fileloc, session)
-        wwwutils.check_dag_warnings(dag.dag_id, session)
-
-        title = "DAG Details"
-        root = request.args.get("root", "")
-
-        states = session.execute(
-            select(TaskInstance.state, sqla.func.count(TaskInstance.dag_id))
-            .where(TaskInstance.dag_id == dag_id)
-            .group_by(TaskInstance.state)
-        ).all()
-
-        active_runs = models.DagRun.find(dag_id=dag_id, state=DagRunState.RUNNING, external_trigger=False)
-
-        tags = session.scalars(select(models.DagTag).where(models.DagTag.dag_id == dag_id)).all()
-
-        # TODO: convert this to a relationship
-        owner_links = session.execute(select(DagOwnerAttributes).filter_by(dag_id=dag_id)).all()
-
-        attrs_to_avoid = [
-            "schedule_datasets",
-            "schedule_dataset_references",
-            "task_outlet_dataset_references",
-            "NUM_DAGS_PER_DAGRUN_QUERY",
-            "serialized_dag",
-            "tags",
-            "default_view",
-            "relative_fileloc",
-            "dag_id",
-            "description",
-            "max_active_runs",
-            "max_active_tasks",
-            "schedule_interval",
-            "owners",
-            "dag_owner_links",
-            "is_paused",
-        ]
-        attrs_to_avoid.extend(wwwutils.get_attr_renderer().keys())
-        dag_model_attrs: list[tuple[str, Any]] = [
-            (attr_name, attr)
-            for attr_name, attr in (
-                (attr_name, getattr(dag_model, attr_name))
-                for attr_name in dir(dag_model)
-                if not attr_name.startswith("_") and attr_name not in attrs_to_avoid
-            )
-            if not callable(attr)
-        ]
-
-        return self.render_template(
-            "airflow/dag_details.html",
-            dag=dag,
-            show_trigger_form_if_no_params=conf.getboolean("webserver", "show_trigger_form_if_no_params"),
-            dag_model=dag_model,
-            title=title,
-            root=root,
-            states=states,
-            State=State,
-            active_runs=active_runs,
-            tags=tags,
-            owner_links=owner_links,
-            dag_model_attrs=dag_model_attrs,
-        )
+        return redirect(url_for("Airflow.grid", dag_id=dag_id))
 
     @expose("/rendered-templates")
     @auth.has_access_dag("GET", DagAccessEntity.TASK_INSTANCE)

tests/www/views/test_views_acl.py

@@ -459,19 +459,6 @@ def test_code_success_for_all_dag_user(client_all_dags_codes, dag_id):
     check_content_in_response(dag_id, resp)
 
 
-def test_dag_details_success(client_all_dags_dagruns):
-    """User without RESOURCE_DAG_CODE can see the page, just not the ID."""
-    url = "dag_details?dag_id=example_bash_operator"
-    resp = client_all_dags_dagruns.get(url, follow_redirects=True)
-    check_content_in_response("DAG Details", resp)
-
-
-def test_dag_details_failure(dag_faker_client):
-    url = "dag_details?dag_id=example_bash_operator"
-    resp = dag_faker_client.get(url, follow_redirects=True)
-    check_content_not_in_response("DAG Details", resp)
-
-
 @pytest.mark.parametrize(
     "dag_id",
     ["example_bash_operator", "example_subdag_operator"],

tests/www/views/test_views_tasks.py

@@ -178,21 +178,6 @@ def client_ti_without_dag_edit(app):
             ["Rendered Template"],
             id="rendered-templates",
         ),
-        pytest.param(
-            "dag_details?dag_id=example_bash_operator",
-            ["DAG Details"],
-            id="dag-details-url-param",
-        ),
-        pytest.param(
-            "dag_details?dag_id=example_subdag_operator.section-1",
-            ["DAG Details"],
-            id="dag-details-subdag-url-param",
-        ),
-        pytest.param(
-            "dags/example_subdag_operator.section-1/details",
-            ["DAG Details"],
-            id="dag-details-subdag",
-        ),
         pytest.param(
             "object/graph_data?dag_id=example_bash_operator",
             ["runme_1"],
@@ -450,22 +435,6 @@ def test_graph_view_without_dag_permission(app, one_dag_perm_user_client):
     check_content_in_response("Access is Denied", resp)
 
 
-def test_dag_details_trigger_origin_dag_details_view(app, admin_client):
-    app.dag_bag.get_dag("test_graph_view").create_dagrun(
-        run_type=DagRunType.SCHEDULED,
-        execution_date=DEFAULT_DATE,
-        data_interval=(DEFAULT_DATE, DEFAULT_DATE),
-        start_date=timezone.utcnow(),
-        state=State.RUNNING,
-    )
-
-    url = "/dags/test_graph_view/details"
-    resp = admin_client.get(url, follow_redirects=True)
-    params = {"origin": "/dags/test_graph_view/details"}
-    href = f"/dags/test_graph_view/trigger?{html.escape(urllib.parse.urlencode(params))}"
-    check_content_in_response(href, resp)
-
-
 def test_last_dagruns(admin_client):
     resp = admin_client.post("last_dagruns", follow_redirects=True)
     check_content_in_response("example_bash_operator", resp)