From c17568e1c3487d5a66a33b03729d8ab05db1e2c0 Mon Sep 17 00:00:00 2001 From: Divyansh Yadav Date: Mon, 5 Oct 2026 18:56:39 +0530 Subject: [PATCH 1/2] Keep get_dag_details and get_pools database queries off the API server event loop --- .../api_fastapi/core_api/datamodels/dags.py | 14 +--------- .../api_fastapi/core_api/datamodels/pools.py | 6 ++-- .../openapi/v2-rest-api-generated.yaml | 12 ++++---- .../core_api/routes/public/dags.py | 8 +++++- .../ui/openapi-gen/requests/schemas.gen.ts | 24 ++++++++-------- .../ui/openapi-gen/requests/types.gen.ts | 5 +--- .../tests/unit/api_fastapi/conftest.py | 28 +++++++++++++++++++ .../core_api/routes/public/test_dags.py | 7 +++++ .../core_api/routes/public/test_pools.py | 8 ++++++ .../airflowctl/api/datamodels/generated.py | 4 +-- 10 files changed, 73 insertions(+), 43 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py index 1030d6e3d4075..17c8c52158f19 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py +++ b/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) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py index 55c7c3ad35e56..8fef6a094ba61 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py +++ b/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 diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 1952f25f7072f..d74464351d5e5 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/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: diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py index ad6531a0ed5c3..7059b9992ba3e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py +++ b/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) diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 8479226b0c839..d6841a98487f3 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/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; diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index ddadbd19f2804..a6e48befe3f62 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/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; }; /** diff --git a/airflow-core/tests/unit/api_fastapi/conftest.py b/airflow-core/tests/unit/api_fastapi/conftest.py index f14b3e6d0bc23..b8fb38057df4f 100644 --- a/airflow-core/tests/unit/api_fastapi/conftest.py +++ b/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.""" diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py index 977f3e08f82b8..47d0706c34a28 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py +++ b/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)) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py index 92c197e74231e..70744d6680ecc 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py +++ b/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( diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index eaedf2055ab4a..1908f0ce9805e 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/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() From 36d57604506ca6f5d038650dcf41c58fd7b5a87d Mon Sep 17 00:00:00 2001 From: Divyansh Yadav Date: Mon, 5 Oct 2026 20:50:53 +0530 Subject: [PATCH 2/2] Validate pools explicitly in get_pools to satisfy mypy --- .../src/airflow/api_fastapi/core_api/routes/public/pools.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/pools.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/pools.py index 07f81b3a0d695..5b8e2098a3761 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/pools.py +++ b/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, )