This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new f812def8776 Allow filtering the Dags list by failed and success runs
in any run state filter (#69875) (#70293)
f812def8776 is described below
commit f812def87762e5bb2481195b8d88f34534e7c7c7
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Thu Jul 23 19:43:49 2026 +0200
Allow filtering the Dags list by failed and success runs in any run state
filter (#69875) (#70293)
(cherry picked from commit a990ce976e7ddba115e3d864d18a84ff58b7e7ed)
Co-authored-by: Yuseok Jo <[email protected]>
---
.../src/airflow/api_fastapi/common/parameters.py | 19 +++++++---------
.../api_fastapi/core_api/openapi/_private_ui.yaml | 6 ++----
.../ui/openapi-gen/queries/ensureQueryData.ts | 2 +-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 2 +-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 2 +-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 2 +-
.../ui/openapi-gen/requests/services.gen.ts | 2 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
.../src/pages/DagsList/DagsFilters/DagsFilters.tsx | 8 +++----
.../api_fastapi/core_api/routes/ui/test_dags.py | 25 +++++++++++-----------
10 files changed, 32 insertions(+), 38 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py
b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index a42cbf9df3a..e033b777772 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -1287,29 +1287,26 @@ QueryPendingActionsFilter =
Annotated[_PendingActionsFilter, Depends(_PendingAct
class _AnyDagRunStateFilter(BaseParam[DagRunState | None]):
"""Filter Dags that have any DagRun in the given state, not only the
latest one."""
- # Only these states have a partial index on dag_run; others would force a
full table scan.
- SUPPORTED_STATES = (DagRunState.QUEUED, DagRunState.RUNNING)
-
def to_orm(self, select: Select) -> Select:
if self.value is None and self.skip_none:
return select
- run_subquery = sql_select(DagRun.dag_id).where(DagRun.state ==
self.value).distinct()
- return select.where(DagModel.dag_id.in_(run_subquery))
+ # EXISTS resolves each Dag via the (dag_id, state) index instead of
scanning every run in the state.
+ has_run_in_state = (
+ sql_select(DagRun.dag_id)
+ .where(DagRun.dag_id == DagModel.dag_id, DagRun.state ==
self.value)
+ .exists()
+ )
+ return select.where(has_run_in_state)
@classmethod
def depends(
cls,
dag_run_state: DagRunState | None = Query(
None,
- description="Filter Dags that have any DagRun in the given state.
Only ``queued`` and ``running`` are supported.",
+ description="Filter Dags that have any DagRun in the given state.",
),
) -> _AnyDagRunStateFilter:
- if dag_run_state is not None and dag_run_state not in
cls.SUPPORTED_STATES:
- raise HTTPException(
- status.HTTP_400_BAD_REQUEST,
- detail=f"dag_run_state only supports {[state.value for state
in cls.SUPPORTED_STATES]}.",
- )
return cls().set_value(dag_run_state)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index e7eb12acfaa..90e221c0703 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -464,11 +464,9 @@ paths:
anyOf:
- $ref: '#/components/schemas/DagRunState'
- type: 'null'
- description: Filter Dags that have any DagRun in the given state.
Only ``queued``
- and ``running`` are supported.
+ description: Filter Dags that have any DagRun in the given state.
title: Dag Run State
- description: Filter Dags that have any DagRun in the given state. Only
``queued``
- and ``running`` are supported.
+ description: Filter Dags that have any DagRun in the given state.
- name: bundle_name
in: query
required: false
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index edcc8e6d4dd..205abe24ec9 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -684,7 +684,7 @@ export const ensureUseDagServiceGetDagTagsData =
(queryClient: QueryClient, { li
* @param data.paused
* @param data.hasImportErrors Filter Dags by having import errors. Only Dags
that have been successfully loaded before will be returned.
* @param data.lastDagRunState
-* @param data.dagRunState Filter Dags that have any DagRun in the given state.
Only ``queued`` and ``running`` are supported.
+* @param data.dagRunState Filter Dags that have any DagRun in the given state.
* @param data.bundleName
* @param data.bundleVersion
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `dag_id,
dag_display_name, next_dagrun, state, start_date, last_run_state,
last_run_start_date`
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index df3d75071d4..52ec99b7f3d 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -684,7 +684,7 @@ export const prefetchUseDagServiceGetDagTags =
(queryClient: QueryClient, { limi
* @param data.paused
* @param data.hasImportErrors Filter Dags by having import errors. Only Dags
that have been successfully loaded before will be returned.
* @param data.lastDagRunState
-* @param data.dagRunState Filter Dags that have any DagRun in the given state.
Only ``queued`` and ``running`` are supported.
+* @param data.dagRunState Filter Dags that have any DagRun in the given state.
* @param data.bundleName
* @param data.bundleVersion
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `dag_id,
dag_display_name, next_dagrun, state, start_date, last_run_state,
last_run_start_date`
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index ab04de50cb1..67e8c4935f2 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -684,7 +684,7 @@ export const useDagServiceGetDagTags = <TData =
Common.DagServiceGetDagTagsDefau
* @param data.paused
* @param data.hasImportErrors Filter Dags by having import errors. Only Dags
that have been successfully loaded before will be returned.
* @param data.lastDagRunState
-* @param data.dagRunState Filter Dags that have any DagRun in the given state.
Only ``queued`` and ``running`` are supported.
+* @param data.dagRunState Filter Dags that have any DagRun in the given state.
* @param data.bundleName
* @param data.bundleVersion
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `dag_id,
dag_display_name, next_dagrun, state, start_date, last_run_state,
last_run_start_date`
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index 4fc85f3bc86..2f96b8cdc97 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -684,7 +684,7 @@ export const useDagServiceGetDagTagsSuspense = <TData =
Common.DagServiceGetDagT
* @param data.paused
* @param data.hasImportErrors Filter Dags by having import errors. Only Dags
that have been successfully loaded before will be returned.
* @param data.lastDagRunState
-* @param data.dagRunState Filter Dags that have any DagRun in the given state.
Only ``queued`` and ``running`` are supported.
+* @param data.dagRunState Filter Dags that have any DagRun in the given state.
* @param data.bundleName
* @param data.bundleVersion
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `dag_id,
dag_display_name, next_dagrun, state, start_date, last_run_state,
last_run_start_date`
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 8bbaed47da9..8b301154b84 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -1959,7 +1959,7 @@ export class DagService {
* @param data.paused
* @param data.hasImportErrors Filter Dags by having import errors. Only
Dags that have been successfully loaded before will be returned.
* @param data.lastDagRunState
- * @param data.dagRunState Filter Dags that have any DagRun in the given
state. Only ``queued`` and ``running`` are supported.
+ * @param data.dagRunState Filter Dags that have any DagRun in the given
state.
* @param data.bundleName
* @param data.bundleVersion
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `dag_id,
dag_display_name, next_dagrun, state, start_date, last_run_state,
last_run_start_date`
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 1c6e5e073b7..47e06ab470d 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
@@ -3446,7 +3446,7 @@ export type GetDagsUiData = {
dagIds?: Array<(string)> | null;
dagRunsLimit?: number;
/**
- * Filter Dags that have any DagRun in the given state. Only ``queued``
and ``running`` are supported.
+ * Filter Dags that have any DagRun in the given state.
*/
dagRunState?: DagRunState | null;
excludeStale?: boolean;
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx
index 9f1863bc120..5a017817719 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx
@@ -45,9 +45,7 @@ const {
type BooleanFilterValue = "all" | "false" | "true";
-const lastRunStates = ["failed", "queued", "running", "success"] as const;
-// Mirrors the backend limit (see _AnyDagRunStateFilter): only running/queued
are indexed on dag_run.
-const anyRunStates = ["queued", "running"] as const;
+const runStates = ["failed", "queued", "running", "success"] as const;
const booleanFilterValues: ReadonlyArray<BooleanFilterValue> = ["all", "true",
"false"];
const toBooleanFilterValue = (
@@ -161,14 +159,14 @@ export const DagsFilters = () => {
dataTestId="dags-last-run-state-filter"
label={translate("filters.lastRunState")}
onChange={handleStateChange}
- states={lastRunStates}
+ states={runStates}
value={state ?? undefined}
/>
<RunStateSelect
dataTestId="dags-any-run-state-filter"
label={translate("filters.anyRunState")}
onChange={handleActiveRunChange}
- states={anyRunStates}
+ states={runStates}
value={activeRunState ?? undefined}
/>
<RequiredActionFilter needsReview={needsReview === "true"}
onToggle={handleNeedsReviewToggle} />
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
index 25e6ac12ff3..04fa6876d47 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
@@ -135,30 +135,31 @@ class TestGetDagRuns(TestPublicDagEndpoint):
previous_run_after = dag_run["run_after"]
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
- def test_dag_run_state_matches_any_run_not_only_latest(self, test_client,
session):
- # Backwards backfill: an older run is still running while the latest
run already finished.
+ @pytest.mark.parametrize("state", ["queued", "running", "failed",
"success"])
+ def test_dag_run_state_matches_any_run_not_only_latest(self, test_client,
session, state):
+ # Give DAG1 an older run in the probed state while its latest run ends
in a
+ # different state, so a latest-run filter misses it but an any-run
filter finds it
+ # (e.g. failure history hidden behind a green latest run).
older_run = session.scalar(
select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
"run_id_1")
)
- older_run.state = DagRunState.RUNNING
+ older_run.state = DagRunState(state)
+ latest_run = session.scalar(
+ select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
"run_id_5")
+ )
+ latest_run.state = DagRunState.FAILED if state == "success" else
DagRunState.SUCCESS
session.commit()
- # last_dag_run_state only looks at the latest run, which is not running
- last_state = test_client.get("/dags", params={"last_dag_run_state":
"running"})
+ # last_dag_run_state only looks at the latest run, which is in a
different state
+ last_state = test_client.get("/dags", params={"last_dag_run_state":
state, "dag_ids": [DAG1_ID]})
assert last_state.status_code == 200
assert [dag["dag_id"] for dag in last_state.json()["dags"]] == []
# dag_run_state matches a Dag that has any run in the state
- any_state = test_client.get("/dags", params={"dag_run_state":
"running"})
+ any_state = test_client.get("/dags", params={"dag_run_state": state,
"dag_ids": [DAG1_ID]})
assert any_state.status_code == 200
assert [dag["dag_id"] for dag in any_state.json()["dags"]] == [DAG1_ID]
- @pytest.mark.parametrize("unsupported_state", ["success", "failed"])
- def test_dag_run_state_rejects_unsupported_states(self, test_client,
unsupported_state):
- # Only running/queued have a partial index; other states would force a
full table scan.
- response = test_client.get("/dags", params={"dag_run_state":
unsupported_state})
- assert response.status_code == 400
-
@pytest.fixture
def setup_hitl_data(self, create_task_instance: TaskInstance, session:
Session):
"""Setup HITL test data for parametrized tests."""
