Ensure total_entries in /api/v1/dags (#43377) (#43429)

* Ensure total_entries in /api/v1/dags

* Test total_entries for /api/v1/dags?fields=...

(cherry picked from commit 19ddb02)

Co-authored-by: xitep <[email protected]>
(cherry picked from commit 6a23692)

airflow/api_connexion/endpoints/dag_endpoint.py

@@ -130,7 +130,7 @@ def get_dags(
 
     try:
         dags_collection_schema = (
-            DAGCollectionSchema(only=[f"dags.{field}" for field in fields])
+            DAGCollectionSchema(only=[f"dags.{field}" for field in fields] + ["total_entries"])
             if fields
             else DAGCollectionSchema()
         )

tests/api_connexion/endpoints/test_dag_endpoint.py

@@ -1288,6 +1288,26 @@ def test_should_return_specified_fields(self):
             for field in fields:
                 assert field in dag
 
+    def test_should_return_specified_fields_and_total_entries(self):
+        total = 4
+        self._create_dag_models(total)
+        self._create_deactivated_dag()
+
+        limit = 2
+        fields = ["dag_id"]
+        response = self.client.get(
+            f"api/v1/dags?limit={limit}&fields={','.join(fields)}", environ_overrides={"REMOTE_USER": "test"}
+        )
+        assert response.status_code == 200
+
+        res_json = response.json
+        assert res_json["total_entries"] == total
+        assert len(res_json["dags"]) == limit
+        for dag in res_json["dags"]:
+            assert len(dag.keys()) == len(fields)
+            for field in fields:
+                assert field in dag
+
     def test_should_respond_400_with_not_exists_fields(self):
         self._create_dag_models(1)
         self._create_deactivated_dag()