[AIP-94] Migrate airflow dags details to the API server

airflow-core/src/airflow/cli/commands/dag_command.py

@@ -610,15 +610,13 @@ def filter_dags_by_bundle(dags: Iterable[DAG], bundle_names: list[str] | None) -
 
 
 @cli_utils.action_cli
+@deprecated_for_airflowctl("airflowctl dags get-details")
 @suppress_logs_and_warning
 @providers_configuration_loaded
-@provide_session
-def dag_details(args, *, session: Session = NEW_SESSION):
+@provide_api_client
+def dag_details(args, api_client: Client = NEW_API_CLIENT):
     """Get DAG details given a DAG id."""
-    dag = DagModel.get_dagmodel(args.dag_id, session=session)
-    if not dag:
-        raise SystemExit(f"DAG: {args.dag_id} does not exist in 'dag' table")
-    dag_detail = DAGResponse.from_orm(dag).model_dump()
+    dag_detail = api_client.dags.get_details(dag_id=args.dag_id).model_dump(mode="json")
 
     if args.output in ["table", "plain"]:
         data = [{"property_name": key, "property_value": value} for key, value in dag_detail.items()]

airflow-core/tests/unit/cli/commands/test_command_deprecations.py

@@ -37,6 +37,7 @@
 DEPRECATED_CLI_COMMANDS = [
     (dag_command.dag_trigger, ["dags", "trigger", "example_dag", "--run-id=x"], "airflowctl dags trigger"),
     (dag_command.dag_delete, ["dags", "delete", "example_dag", "--yes"], "airflowctl dags delete"),
+    (dag_command.dag_details, ["dags", "details", "example_dag"], "airflowctl dags get-details"),
     (pool_command.pool_list, ["pools", "list"], "airflowctl pools list"),
     (pool_command.pool_get, ["pools", "get", "foo"], "airflowctl pools get"),
     (pool_command.pool_set, ["pools", "set", "foo", "1", "desc"], "airflowctl pools create"),

airflow-core/tests/unit/cli/commands/test_dag_command.py

@@ -327,22 +327,6 @@ def test_cli_report(self, stdout_capture):
         assert any(item["file"].endswith("example_complex.py") for item in data)
         assert any("example_complex" in item["dags"] for item in data)
 
-    def test_cli_get_dag_details(self, stdout_capture):
-        args = self.parser.parse_args(["dags", "details", "example_complex", "--output", "yaml"])
-        with stdout_capture as temp_stdout:
-            dag_command.dag_details(args)
-            out = temp_stdout.getvalue()
-
-        # Check if DAG Details field are present
-        for field in dag_command.DAG_DETAIL_FIELDS:
-            assert field in out
-
-        # Check if identifying values are present
-        dag_details_values = ["airflow", "airflow/example_dags/example_complex.py", "16", "example_complex"]
-
-        for value in dag_details_values:
-            assert value in out
-
     def test_cli_list_dags(self, stdout_capture):
         args = self.parser.parse_args(["dags", "list", "--output", "json"])
         with stdout_capture as temp_stdout:
@@ -1938,6 +1922,46 @@ def _default_trigger_response(self, mock_cli_api_client):
             "dag_run_id": "test_run",
         }
 
+    def test_get_dag_details(self, mock_cli_api_client, stdout_capture):
+        mock_cli_api_client.dags.get_details.return_value.model_dump.return_value = {
+            "dag_id": "example_complex",
+            "fileloc": "airflow/example_dags/example_complex.py",
+            "owners": ["airflow"],
+            "max_active_tasks": 16,
+        }
+        args = self.parser.parse_args(["dags", "details", "example_complex", "--output", "yaml"])
+
+        with stdout_capture as temp_stdout:
+            dag_command.dag_details(args)
+            out = temp_stdout.getvalue()
+
+        mock_cli_api_client.dags.get_details.assert_called_once_with(dag_id="example_complex")
+        mock_cli_api_client.dags.get_details.return_value.model_dump.assert_called_once_with(mode="json")
+        for value in ["airflow", "airflow/example_dags/example_complex.py", "16", "example_complex"]:
+            assert value in out
+
+    def test_get_dag_details_plain_output(self, mock_cli_api_client, stdout_capture):
+        mock_cli_api_client.dags.get_details.return_value.model_dump.return_value = {
+            "dag_id": "example_complex",
+            "is_paused": False,
+        }
+        args = self.parser.parse_args(["dags", "details", "example_complex", "--output", "plain"])
+
+        with stdout_capture as temp_stdout:
+            dag_command.dag_details(args)
+            out = temp_stdout.getvalue()
+
+        assert "property_name" in out
+        assert "property_value" in out
+        assert "dag_id" in out
+        assert "example_complex" in out
+
+    def test_get_dag_details_missing(self, mock_cli_api_client):
+        mock_cli_api_client.dags.get_details.side_effect = _server_error(404)
+
+        with pytest.raises(ServerResponseError):
+            dag_command.dag_details(self.parser.parse_args(["dags", "details", "does_not_exist"]))
+
     def test_trigger_dag(self, mock_cli_api_client):
         dag_command.dag_trigger(
             self.parser.parse_args(