Keep get_dag_details and get_pools database queries off the API server event loop

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py

@@ -40,7 +40,6 @@
 from airflow.api_fastapi.core_api.datamodels.dag_tags import DagTagResponse
 from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
 from airflow.configuration import conf
-from airflow.models.dag_version import DagVersion
 from airflow.utils.state import DagSchedulingState
 from airflow.utils.types import DagRunType
 
@@ -228,6 +227,7 @@ class DAGDetailsResponse(DAGResponse):
     is_favorite: bool = False
     active_runs_count: int = 0
     team_name: str | None = None
+    latest_dag_version: DagVersionResponse | None
 
     @field_validator("timezone", mode="before")
     @classmethod
@@ -294,15 +294,3 @@ def concurrency(self) -> int:
         Deprecated: Use max_active_tasks instead.
         """
         return self.max_active_tasks
-
-    # Mypy issue https://github.com/python/mypy/issues/1362
-    @computed_field  # type: ignore[prop-decorator]
-    @property
-    def latest_dag_version(self) -> DagVersionResponse | None:
-        """Return the latest DagVersion."""
-        latest_dag_version = DagVersion.get_latest_version(
-            self.dag_id, load_dag_model=True, load_bundle_model=True
-        )
-        if latest_dag_version is None:
-            return latest_dag_version
-        return DagVersionResponse.model_validate(latest_dag_version)

airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py

@@ -17,7 +17,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable, Iterable
+from collections.abc import Callable
 from typing import Annotated
 
 from pydantic import BeforeValidator, Field, model_validator
@@ -71,7 +71,9 @@ class PoolResponse(BasePool):
 class PoolCollectionResponse(BaseModel):
     """Pool Collection serializer for responses."""
 
-    pools: Iterable[PoolResponse]
+    # Not Iterable: pydantic validates Iterable items lazily during serialization, which would
+    # run the slot queries of each pool on the event loop.
+    pools: list[PoolResponse]
     total_entries: int
 
 

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -13764,6 +13764,10 @@ components:
           - type: string
           - type: 'null'
           title: Team Name
+        latest_dag_version:
+          anyOf:
+          - $ref: '#/components/schemas/DagVersionResponse'
+          - type: 'null'
         is_backfillable:
           type: boolean
           title: Is Backfillable
@@ -13783,12 +13787,6 @@ components:
             Deprecated: Use max_active_tasks instead.'
           deprecated: true
           readOnly: true
-        latest_dag_version:
-          anyOf:
-          - $ref: '#/components/schemas/DagVersionResponse'
-          - type: 'null'
-          description: Return the latest DagVersion.
-          readOnly: true
       type: object
       required:
       - dag_id
@@ -13832,10 +13830,10 @@ components:
       - timezone
       - last_parsed
       - default_args
+      - latest_dag_version
       - is_backfillable
       - file_token
       - concurrency
-      - latest_dag_version
       title: DAGDetailsResponse
       description: Specific serializer for Dag Details responses.
     DAGPatchBody:

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py

@@ -80,6 +80,7 @@
 from airflow.exceptions import AirflowException, DagNotFound
 from airflow.models import DagModel
 from airflow.models.dag_favorite import DagFavorite
+from airflow.models.dag_version import DagVersion
 from airflow.models.dagrun import DagRun
 from airflow.utils.sqlalchemy import with_row_locks
 from airflow.utils.state import DagRunState, DagSchedulingState
@@ -270,9 +271,14 @@ def get_dag_details(
         or 0
     )
 
-    # Add is_favorite and active_runs_count fields to the Dag model
+    latest_dag_version = DagVersion.get_latest_version(
+        dag_id, load_dag_model=True, load_bundle_model=True, session=session
+    )
+
+    # Add is_favorite, active_runs_count and latest_dag_version fields to the Dag model
     setattr(dag_model, "is_favorite", is_favorite)
     setattr(dag_model, "active_runs_count", active_runs_count)
+    setattr(dag_model, "latest_dag_version", latest_dag_version)
 
     return DAGDetailsResponse.model_validate(dag_model)
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/pools.py

@@ -122,7 +122,7 @@ def get_pools(
     pools = session.scalars(pools_select)
 
     return PoolCollectionResponse(
-        pools=pools,
+        pools=[PoolResponse.model_validate(pool) for pool in pools],
         total_entries=total_entries,
     )
 

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -3472,6 +3472,16 @@ export const $DAGDetailsResponse = {
             ],
             title: 'Team Name'
         },
+        latest_dag_version: {
+            anyOf: [
+                {
+                    '$ref': '#/components/schemas/DagVersionResponse'
+                },
+                {
+                    type: 'null'
+                }
+            ]
+        },
         is_backfillable: {
             type: 'boolean',
             title: 'Is Backfillable',
@@ -3492,22 +3502,10 @@ export const $DAGDetailsResponse = {
 Deprecated: Use max_active_tasks instead.`,
             deprecated: true,
             readOnly: true
-        },
-        latest_dag_version: {
-            anyOf: [
-                {
-                    '$ref': '#/components/schemas/DagVersionResponse'
-                },
-                {
-                    type: 'null'
-                }
-            ],
-            description: 'Return the latest DagVersion.',
-            readOnly: true
         }
     },
     type: 'object',
-    required: ['dag_id', 'dag_display_name', 'is_paused', 'is_stale', 'last_parsed_time', 'last_parse_duration', 'last_expired', 'bundle_name', 'bundle_version', 'relative_fileloc', 'fileloc', 'description', 'timetable_summary', 'timetable_description', 'timetable_partitioned', 'timetable_periodic', 'tags', 'max_active_tasks', 'max_active_runs', 'max_consecutive_failed_dag_runs', 'has_task_concurrency_limits', 'has_import_errors', 'next_dagrun_logical_date', 'next_dagrun_data_interval_start', 'next_dagrun_data_interval_end', 'next_dagrun_run_after', 'allowed_run_types', 'owners', 'catchup', 'dag_run_timeout', 'asset_expression', 'doc_md', 'start_date', 'end_date', 'is_paused_upon_creation', 'params', 'render_template_as_native_obj', 'template_search_path', 'timezone', 'last_parsed', 'default_args', 'is_backfillable', 'file_token', 'concurrency', 'latest_dag_version'],
+    required: ['dag_id', 'dag_display_name', 'is_paused', 'is_stale', 'last_parsed_time', 'last_parse_duration', 'last_expired', 'bundle_name', 'bundle_version', 'relative_fileloc', 'fileloc', 'description', 'timetable_summary', 'timetable_description', 'timetable_partitioned', 'timetable_periodic', 'tags', 'max_active_tasks', 'max_active_runs', 'max_consecutive_failed_dag_runs', 'has_task_concurrency_limits', 'has_import_errors', 'next_dagrun_logical_date', 'next_dagrun_data_interval_start', 'next_dagrun_data_interval_end', 'next_dagrun_run_after', 'allowed_run_types', 'owners', 'catchup', 'dag_run_timeout', 'asset_expression', 'doc_md', 'start_date', 'end_date', 'is_paused_upon_creation', 'params', 'render_template_as_native_obj', 'template_search_path', 'timezone', 'last_parsed', 'default_args', 'latest_dag_version', 'is_backfillable', 'file_token', 'concurrency'],
     title: 'DAGDetailsResponse',
     description: 'Specific serializer for Dag Details responses.'
 } as const;

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -973,6 +973,7 @@ export type DAGDetailsResponse = {
     is_favorite?: boolean;
     active_runs_count?: number;
     team_name?: string | null;
+    latest_dag_version: DagVersionResponse | null;
     /**
      * Whether this Dag's schedule supports backfilling.
      */
@@ -988,10 +989,6 @@ export type DAGDetailsResponse = {
      * @deprecated
      */
     readonly concurrency: number;
-    /**
-     * Return the latest DagVersion.
-     */
-    readonly latest_dag_version: DagVersionResponse | null;
 };
 
 /**

airflow-core/tests/unit/api_fastapi/conftest.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import asyncio
 import datetime
 import os
 from typing import TYPE_CHECKING
@@ -26,6 +27,8 @@
 from fastapi import FastAPI
 from fastapi.routing import Mount
 from fastapi.testclient import TestClient
+from sqlalchemy import event
+from sqlalchemy.engine import Engine
 
 from airflow.api_fastapi.app import create_app
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
@@ -205,6 +208,31 @@ def deny_dag_edit_access():
         yield mock_is_authorized_dag
 
 
+@pytest.fixture
+def event_loop_queries():
+    """
+    Collect the SQL statements executed on the event loop thread.
+
+    FastAPI serializes responses on the event loop, so a statement collected here blocks the whole
+    API server while it waits for a database connection.
+    """
+    statements: list[str] = []
+
+    def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
+        # Only the event loop thread has a running loop; threadpool workers raise here.
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        statements.append(statement)
+
+    event.listen(Engine, "before_cursor_execute", before_cursor_execute)
+    try:
+        yield statements
+    finally:
+        event.remove(Engine, "before_cursor_execute", before_cursor_execute)
+
+
 @pytest.fixture
 def client(request):
     """This fixture is more flexible than test_client, as it allows to specify which apps to include."""

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py

@@ -1840,6 +1840,13 @@ def test_dag_details_team_name_none_without_multi_team(self, test_client):
         assert response.status_code == 200
         assert response.json()["team_name"] is None
 
+    def test_dag_details_should_not_query_on_event_loop(self, test_client, event_loop_queries):
+        """Regression for #74239: the latest Dag version must not be queried during serialization."""
+        response = test_client.get(f"/dags/{DAG2_ID}/details")
+        assert response.status_code == 200
+        assert response.json()["latest_dag_version"]["version_number"] == 1
+        assert event_loop_queries == []
+
     @conf_vars({("core", "multi_team"): "True"})
     def test_dag_details_includes_team_name(self, session, test_client):
         original_bundle_name = session.scalar(select(DagModel.bundle_name).where(DagModel.dag_id == DAG1_ID))

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py

@@ -272,6 +272,14 @@ def test_get_pools_with_unlimited_pool_should_respond_200(self, test_client, ses
         assert pools["unlimited_pool"]["slots"] == -1
         assert pools["unlimited_pool"]["open_slots"] == -1
 
+    def test_should_not_query_on_event_loop(self, test_client, event_loop_queries):
+        """Regression for #74239: the slot counts of each pool must not be queried during serialization."""
+        self.create_pools()
+        response = test_client.get("/pools")
+        assert response.status_code == 200
+        assert response.json()["pools"][1]["open_slots"] == POOL1_SLOT
+        assert event_loop_queries == []
+
 
 class TestPatchPool(TestPoolsEndpoint):
     @pytest.mark.parametrize(

airflow-ctl/src/airflowctl/api/datamodels/generated.py

@@ -3204,6 +3204,7 @@ class DAGDetailsResponse(BaseModel):
     is_favorite: Annotated[bool | None, Field(title="Is Favorite")] = False
     active_runs_count: Annotated[int | None, Field(title="Active Runs Count")] = 0
     team_name: Annotated[str | None, Field(title="Team Name")] = None
+    latest_dag_version: DagVersionResponse | None
     is_backfillable: Annotated[
         bool, Field(description="Whether this Dag's schedule supports backfilling.", title="Is Backfillable")
     ]
@@ -3216,9 +3217,6 @@ class DAGDetailsResponse(BaseModel):
             title="Concurrency",
         ),
     ]
-    latest_dag_version: Annotated[
-        DagVersionResponse | None, Field(description="Return the latest DagVersion.")
-    ]
 
 
 AssetExpressionAll.model_rebuild()