This is an automated email from the ASF dual-hosted git repository.
guan404ming pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e05d6cd3215 Add support for filtering Dags by any DagRun state (#68657)
e05d6cd3215 is described below
commit e05d6cd3215f3ee71388cc5cc91ef058e2d016c2
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Thu Jul 9 16:00:14 2026 +0800
Add support for filtering Dags by any DagRun state (#68657)
* Add support for filtering Dags by any DagRun state
* Update
airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx
Co-authored-by: Pierre Jeambrun <[email protected]>
* Switched the fixed width to fit-content
---------
Co-authored-by: Pierre Jeambrun <[email protected]>
---
.../src/airflow/api_fastapi/common/parameters.py | 31 ++++++++
.../api_fastapi/core_api/openapi/_private_ui.yaml | 12 +++
.../airflow/api_fastapi/core_api/routes/ui/dags.py | 3 +
.../api_fastapi/core_api/routes/ui/dashboard.py | 44 +++++------
.../src/airflow/ui/openapi-gen/queries/common.ts | 5 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 6 +-
.../ui/openapi-gen/requests/services.gen.ts | 2 +
.../airflow/ui/openapi-gen/requests/types.gen.ts | 4 +
.../airflow/ui/public/i18n/locales/en/dags.json | 2 +
.../src/airflow/ui/src/constants/searchParams.ts | 1 +
.../DagsList/DagsFilters/DagsFilters.test.tsx | 7 +-
.../src/pages/DagsList/DagsFilters/DagsFilters.tsx | 48 ++++++++----
.../pages/DagsList/DagsFilters/RunStateSelect.tsx | 87 ++++++++++++++++++++++
.../pages/DagsList/DagsFilters/StateFilters.tsx | 75 -------------------
.../ui/src/pages/DagsList/DagsList.test.tsx | 14 ++--
.../src/airflow/ui/src/pages/DagsList/DagsList.tsx | 3 +
.../airflow/ui/src/pages/Dashboard/Stats/Stats.tsx | 4 +-
.../src/airflow/ui/src/queries/useDags.tsx | 3 +
.../src/airflow/ui/tests/e2e/pages/DagsPage.ts | 26 +++----
.../airflow/ui/tests/e2e/specs/dags-list.spec.ts | 7 +-
.../ui/tests/e2e/specs/home-dashboard.spec.ts | 2 +-
.../api_fastapi/core_api/routes/ui/test_dags.py | 26 +++++++
.../core_api/routes/ui/test_dashboard.py | 2 +-
26 files changed, 273 insertions(+), 159 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py
b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index 7b235f07de3..c3a543a2b90 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -1338,11 +1338,42 @@ class _PendingActionsFilter(BaseParam[bool]):
QueryPendingActionsFilter = Annotated[_PendingActionsFilter,
Depends(_PendingActionsFilter.depends)]
+
+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))
+
+ @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.",
+ ),
+ ) -> _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)
+
+
# DagRun
QueryLastDagRunStateFilter = Annotated[
FilterParam[DagRunState | None],
Depends(filter_param_factory(DagRun.state, DagRunState | None,
filter_name="last_dag_run_state")),
]
+QueryAnyDagRunStateFilter = Annotated[_AnyDagRunStateFilter,
Depends(_AnyDagRunStateFilter.depends)]
def _transform_dag_run_states(states: Iterable[str] | None) ->
list[DagRunState | None] | None:
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 173f231a098..3defcad70f9 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
@@ -473,6 +473,18 @@ paths:
- $ref: '#/components/schemas/DagRunState'
- type: 'null'
title: Last Dag Run State
+ - name: dag_run_state
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - $ref: '#/components/schemas/DagRunState'
+ - type: 'null'
+ description: Filter Dags that have any DagRun in the given state.
Only ``queued``
+ and ``running`` are supported.
+ title: Dag Run State
+ description: Filter Dags that have any DagRun in the given state. Only
``queued``
+ and ``running`` are supported.
- name: bundle_name
in: query
required: false
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
index 15515a92bb8..38d0c93cef8 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
@@ -32,6 +32,7 @@ from airflow.api_fastapi.common.db.dags import
generate_dag_with_latest_run_quer
from airflow.api_fastapi.common.parameters import (
FilterOptionEnum,
FilterParam,
+ QueryAnyDagRunStateFilter,
QueryAssetDependencyFilter,
QueryBundleNameFilter,
QueryBundleVersionFilter,
@@ -107,6 +108,7 @@ def get_dags(
paused: QueryPausedFilter,
has_import_errors: QueryHasImportErrorsFilter,
last_dag_run_state: QueryLastDagRunStateFilter,
+ dag_run_state: QueryAnyDagRunStateFilter,
bundle_name: QueryBundleNameFilter,
bundle_version: QueryBundleVersionFilter,
order_by: Annotated[
@@ -152,6 +154,7 @@ def get_dags(
tags,
owners,
last_dag_run_state,
+ dag_run_state,
is_favorite,
has_asset_schedule,
asset_dependency,
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
index 612d5c7c35c..0498082a4ed 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
@@ -129,16 +129,6 @@ def dag_stats(
"""Return basic Dag stats with counts of Dags in various states."""
permitted_dag_ids = cast("set[str]", readable_dags_filter.value)
- # Active Dags need another query from DagModel, as a Dag may not have any
runs but still be active
- active_count_query = (
- select(func.count())
- .select_from(DagModel)
- .where(DagModel.is_stale == false())
- .where(DagModel.is_paused == false())
- .where(DagModel.dag_id.in_(permitted_dag_ids))
- )
- active_count = session.execute(active_count_query).scalar_one()
-
latest_state = (
select(DagRun.state)
.where(DagRun.dag_id == DagModel.dag_id,
DagRun.logical_date.is_not(None))
@@ -147,26 +137,30 @@ def dag_stats(
.correlate(DagModel)
.scalar_subquery()
)
- latest_runs_subq = (
- select(latest_state.label("state"))
+ dag_counts_query = (
+ select(
+ func.coalesce(func.sum(case((DagModel.is_paused == false(), 1))),
0).label("active"),
+ func.coalesce(func.sum(case((latest_state == DagRunState.FAILED,
1))), 0).label("failed"),
+ )
.select_from(DagModel)
.where(DagModel.is_stale == false())
.where(DagModel.dag_id.in_(permitted_dag_ids))
- .subquery()
)
- combined_runs_query = select(
- func.coalesce(func.sum(case((latest_runs_subq.c.state ==
DagRunState.FAILED, 1))), 0).label("failed"),
- func.coalesce(func.sum(case((latest_runs_subq.c.state ==
DagRunState.RUNNING, 1))), 0).label(
- "running"
- ),
- func.coalesce(func.sum(case((latest_runs_subq.c.state ==
DagRunState.QUEUED, 1))), 0).label("queued"),
- ).select_from(latest_runs_subq)
+ dag_counts = session.execute(dag_counts_query).one()
- counts = session.execute(combined_runs_query).one()
+ active_states_query = (
+ select(DagRun.state, func.count(func.distinct(DagRun.dag_id)))
+ .join(DagModel, DagModel.dag_id == DagRun.dag_id)
+ .where(DagModel.is_stale == false())
+ .where(DagRun.dag_id.in_(permitted_dag_ids))
+ .where(DagRun.state.in_([DagRunState.RUNNING, DagRunState.QUEUED]))
+ .group_by(DagRun.state)
+ )
+ active_counts = {state: count for state, count in
session.execute(active_states_query)}
return DashboardDagStatsResponse(
- active_dag_count=active_count,
- failed_dag_count=counts.failed,
- running_dag_count=counts.running,
- queued_dag_count=counts.queued,
+ active_dag_count=dag_counts.active,
+ failed_dag_count=dag_counts.failed,
+ running_dag_count=active_counts.get(DagRunState.RUNNING, 0),
+ queued_dag_count=active_counts.get(DagRunState.QUEUED, 0),
)
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index 769f3eb49fa..2d6dc7ffe7d 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -327,7 +327,7 @@ export const UseDagServiceGetDagTagsKeyFn = ({ limit,
offset, orderBy, tagNamePa
export type DagServiceGetDagsUiDefaultResponse = Awaited<ReturnType<typeof
DagService.getDagsUi>>;
export type DagServiceGetDagsUiQueryResult<TData =
DagServiceGetDagsUiDefaultResponse, TError = unknown> = UseQueryResult<TData,
TError>;
export const useDagServiceGetDagsUiKey = "DagServiceGetDagsUi";
-export const UseDagServiceGetDagsUiKeyFn = ({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode }:
{
+export const UseDagServiceGetDagsUiKeyFn = ({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState,
excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode }:
{
assetDependency?: string;
bundleName?: string;
bundleVersion?: string;
@@ -337,6 +337,7 @@ export const UseDagServiceGetDagsUiKeyFn = ({
assetDependency, bundleName, bundl
dagIdPrefixPattern?: string;
dagIds?: string[];
dagRunsLimit?: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
hasAssetSchedule?: boolean;
hasImportErrors?: boolean;
@@ -350,7 +351,7 @@ export const UseDagServiceGetDagsUiKeyFn = ({
assetDependency, bundleName, bundl
paused?: boolean;
tags?: string[];
tagsMatchMode?: "any" | "all";
-} = {}, queryKey?: Array<unknown>) => [useDagServiceGetDagsUiKey, ...(queryKey
?? [{ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }])];
+} = {}, queryKey?: Array<unknown>) => [useDagServiceGetDagsUiKey, ...(queryKey
?? [{ assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }])];
export type DagServiceGetLatestRunInfoDefaultResponse =
Awaited<ReturnType<typeof DagService.getLatestRunInfo>>;
export type DagServiceGetLatestRunInfoQueryResult<TData =
DagServiceGetLatestRunInfoDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useDagServiceGetLatestRunInfoKey = "DagServiceGetLatestRunInfo";
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 712d2ee99dd..6eba7e9a9db 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -665,6 +665,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.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`
@@ -675,7 +676,7 @@ export const ensureUseDagServiceGetDagTagsData =
(queryClient: QueryClient, { li
* @returns DAGWithLatestDagRunsCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseDagServiceGetDagsUiData = (queryClient: QueryClient, {
assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }: {
+export const ensureUseDagServiceGetDagsUiData = (queryClient: QueryClient, {
assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }: {
assetDependency?: string;
bundleName?: string;
bundleVersion?: string;
@@ -685,6 +686,7 @@ export const ensureUseDagServiceGetDagsUiData =
(queryClient: QueryClient, { ass
dagIdPrefixPattern?: string;
dagIds?: string[];
dagRunsLimit?: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
hasAssetSchedule?: boolean;
hasImportErrors?: boolean;
@@ -698,7 +700,7 @@ export const ensureUseDagServiceGetDagsUiData =
(queryClient: QueryClient, { ass
paused?: boolean;
tags?: string[];
tagsMatchMode?: "any" | "all";
-} = {}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode
}), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName,
bundleVersion, dagDispla [...]
+} = {}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState,
excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode
}), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName,
bundleVersi [...]
/**
* Get Latest Run Info
* Get latest run.
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 1614a33ac5d..037f6d01a39 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -665,6 +665,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.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`
@@ -675,7 +676,7 @@ export const prefetchUseDagServiceGetDagTags =
(queryClient: QueryClient, { limi
* @returns DAGWithLatestDagRunsCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseDagServiceGetDagsUi = (queryClient: QueryClient, {
assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }: {
+export const prefetchUseDagServiceGetDagsUi = (queryClient: QueryClient, {
assetDependency, bundleName, bundleVersion, dagDisplayNamePattern,
dagDisplayNamePrefixPattern, dagIdPattern, dagIdPrefixPattern, dagIds,
dagRunsLimit, dagRunState, excludeStale, hasAssetSchedule, hasImportErrors,
hasPendingActions, isFavorite, lastDagRunState, limit, offset, orderBy, owners,
paused, tags, tagsMatchMode }: {
assetDependency?: string;
bundleName?: string;
bundleVersion?: string;
@@ -685,6 +686,7 @@ export const prefetchUseDagServiceGetDagsUi = (queryClient:
QueryClient, { asset
dagIdPrefixPattern?: string;
dagIds?: string[];
dagRunsLimit?: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
hasAssetSchedule?: boolean;
hasImportErrors?: boolean;
@@ -698,7 +700,7 @@ export const prefetchUseDagServiceGetDagsUi = (queryClient:
QueryClient, { asset
paused?: boolean;
tags?: string[];
tagsMatchMode?: "any" | "all";
-} = {}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode
}), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName,
bundleVersion, dagDisplayN [...]
+} = {}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState,
excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode
}), queryFn: () => DagService.getDagsUi({ assetDependency, bundleName,
bundleVersion [...]
/**
* Get Latest Run Info
* Get latest run.
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 9f326c03b1b..09ddcdd3837 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -665,6 +665,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.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`
@@ -675,7 +676,7 @@ export const useDagServiceGetDagTags = <TData =
Common.DagServiceGetDagTagsDefau
* @returns DAGWithLatestDagRunsCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagServiceGetDagsUi = <TData =
Common.DagServiceGetDagsUiDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ assetDependency, bundleName, bundleVersion,
dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern,
dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale, hasAssetSchedule,
hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset,
orderBy, owners, paused, tags, tagsMatchMode }: {
+export const useDagServiceGetDagsUi = <TData =
Common.DagServiceGetDagsUiDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ assetDependency, bundleName, bundleVersion,
dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern,
dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode }:
{
assetDependency?: string;
bundleName?: string;
bundleVersion?: string;
@@ -685,6 +686,7 @@ export const useDagServiceGetDagsUi = <TData =
Common.DagServiceGetDagsUiDefault
dagIdPrefixPattern?: string;
dagIds?: string[];
dagRunsLimit?: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
hasAssetSchedule?: boolean;
hasImportErrors?: boolean;
@@ -698,7 +700,7 @@ export const useDagServiceGetDagsUi = <TData =
Common.DagServiceGetDagsUiDefault
paused?: boolean;
tags?: string[];
tagsMatchMode?: "any" | "all";
-} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode },
qu [...]
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState,
excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMa [...]
/**
* Get Latest Run Info
* Get latest run.
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 d2d51a73acd..e3ec8836309 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -665,6 +665,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.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`
@@ -675,7 +676,7 @@ export const useDagServiceGetDagTagsSuspense = <TData =
Common.DagServiceGetDagT
* @returns DAGWithLatestDagRunsCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagServiceGetDagsUiSuspense = <TData =
Common.DagServiceGetDagsUiDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ assetDependency, bundleName, bundleVersion,
dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern,
dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale, hasAssetSchedule,
hasImportErrors, hasPendingActions, isFavorite, lastDagRunState, limit, offset,
orderBy, owners, paused, tags, tagsMatchMode }: {
+export const useDagServiceGetDagsUiSuspense = <TData =
Common.DagServiceGetDagsUiDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ assetDependency, bundleName, bundleVersion,
dagDisplayNamePattern, dagDisplayNamePrefixPattern, dagIdPattern,
dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMode }:
{
assetDependency?: string;
bundleName?: string;
bundleVersion?: string;
@@ -685,6 +686,7 @@ export const useDagServiceGetDagsUiSuspense = <TData =
Common.DagServiceGetDagsU
dagIdPrefixPattern?: string;
dagIds?: string[];
dagRunsLimit?: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
hasAssetSchedule?: boolean;
hasImportErrors?: boolean;
@@ -698,7 +700,7 @@ export const useDagServiceGetDagsUiSuspense = <TData =
Common.DagServiceGetDagsU
paused?: boolean;
tags?: string[];
tagsMatchMode?: "any" | "all";
-} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, excludeStale,
hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags, tagsMatchMo [...]
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagsUiKeyFn({ assetDependency, bundleName,
bundleVersion, dagDisplayNamePattern, dagDisplayNamePrefixPattern,
dagIdPattern, dagIdPrefixPattern, dagIds, dagRunsLimit, dagRunState,
excludeStale, hasAssetSchedule, hasImportErrors, hasPendingActions, isFavorite,
lastDagRunState, limit, offset, orderBy, owners, paused, tags [...]
/**
* Get Latest Run Info
* Get latest run.
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 06842b72e7d..78241796546 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
@@ -1923,6 +1923,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.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`
@@ -1953,6 +1954,7 @@ export class DagService {
paused: data.paused,
has_import_errors: data.hasImportErrors,
last_dag_run_state: data.lastDagRunState,
+ dag_run_state: data.dagRunState,
bundle_name: data.bundleName,
bundle_version: data.bundleVersion,
order_by: data.orderBy,
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 103c071ec87..e1bf23b1602 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
@@ -3490,6 +3490,10 @@ export type GetDagsUiData = {
dagIdPrefixPattern?: string | null;
dagIds?: Array<(string)> | null;
dagRunsLimit?: number;
+ /**
+ * Filter Dags that have any DagRun in the given state. Only ``queued``
and ``running`` are supported.
+ */
+ dagRunState?: DagRunState | null;
excludeStale?: boolean;
/**
* Filter Dags with asset-based scheduling
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json
index 990d6c7848f..c8b195c93b8 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json
@@ -10,11 +10,13 @@
"filters": {
"allRunTypes": "All Run Types",
"allStates": "All States",
+ "anyRunState": "Any run",
"favorite": {
"all": "All",
"favorite": "Favorite",
"unfavorite": "Unfavorite"
},
+ "lastRunState": "Last run",
"paused": {
"active": "Active",
"all": "All",
diff --git a/airflow-core/src/airflow/ui/src/constants/searchParams.ts
b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
index f3583d449ce..b6f4d88e3d0 100644
--- a/airflow-core/src/airflow/ui/src/constants/searchParams.ts
+++ b/airflow-core/src/airflow/ui/src/constants/searchParams.ts
@@ -31,6 +31,7 @@ export enum SearchParamsKeys {
DAG_DISPLAY_NAME_PATTERN = "dag_display_name_pattern",
DAG_ID = "dag_id",
DAG_ID_PATTERN = "dag_id_pattern",
+ DAG_RUN_STATE = "dag_run_state",
DAG_VERSION = "dag_version",
DAG_VIEW = "view",
DEADLINE_TIME_GTE = "deadline_time_gte",
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx
index 4a28ebe8087..aac7b895ae4 100644
---
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx
+++
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx
@@ -52,11 +52,8 @@ describe("Paused filter with hide_paused_dags_by_default
enabled", () => {
await waitFor(() =>
expect(screen.getByText("tutorial_taskflow_api_success")).toBeInTheDocument());
expect(screen.queryByText("paused_dag")).not.toBeInTheDocument();
- // There are two "All" buttons (StateFilters and PausedFilter).
- // The second one belongs to PausedFilter.
- const allButtons = screen.getAllByText("filters.paused.all");
-
- allButtons[1]?.click();
+ // PausedFilter is the only filter using the "All" (filters.paused.all)
label.
+ screen.getByText("filters.paused.all").click();
await waitFor(() =>
expect(screen.getByText("paused_dag")).toBeInTheDocument());
expect(screen.getByText("tutorial_taskflow_api_success")).toBeInTheDocument();
});
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 a8e018bc121..9f1863bc120 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
@@ -16,9 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { Box, HStack } from "@chakra-ui/react";
+import { HStack } from "@chakra-ui/react";
import type { MultiValue } from "chakra-react-select";
import { useState } from "react";
+import { useTranslation } from "react-i18next";
import { useSearchParams } from "react-router-dom";
import { useTableURLState } from "src/components/DataTable/useTableUrlState";
@@ -30,10 +31,11 @@ import { useTagFilter } from "../useTagFilter";
import { FavoriteFilter } from "./FavoriteFilter";
import { PausedFilter } from "./PausedFilter";
import { RequiredActionFilter } from "./RequiredActionFilter";
-import { StateFilters } from "./StateFilters";
+import { RunStateSelect } from "./RunStateSelect";
import { TagFilter } from "./TagFilter";
const {
+ DAG_RUN_STATE: DAG_RUN_STATE_PARAM,
FAVORITE: FAVORITE_PARAM,
LAST_DAG_RUN_STATE: LAST_DAG_RUN_STATE_PARAM,
NEEDS_REVIEW: NEEDS_REVIEW_PARAM,
@@ -41,15 +43,13 @@ const {
PAUSED: PAUSED_PARAM,
}: SearchParamsKeysType = SearchParamsKeys;
-type StateValue = "all" | "failed" | "queued" | "running" | "success";
type BooleanFilterValue = "all" | "false" | "true";
-const stateValues: ReadonlyArray<StateValue> = ["failed", "queued", "running",
"success"];
+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 booleanFilterValues: ReadonlyArray<BooleanFilterValue> = ["all", "true",
"false"];
-const toStateValue = (value: string | null): StateValue =>
- stateValues.includes(value as StateValue) ? (value as StateValue) : "all";
-
const toBooleanFilterValue = (
value: string | null,
defaultValue: BooleanFilterValue = "all",
@@ -58,12 +58,14 @@ const toBooleanFilterValue = (
export const DagsFilters = () => {
const [searchParams, setSearchParams] = useSearchParams();
+ const { t: translate } = useTranslation("dags");
const { selectedTags, setSelectedTags, setTagFilterMode, tagFilterMode } =
useTagFilter();
const showPaused = searchParams.get(PAUSED_PARAM);
const showFavorites = searchParams.get(FAVORITE_PARAM);
const needsReview = searchParams.get(NEEDS_REVIEW_PARAM);
const state = searchParams.get(LAST_DAG_RUN_STATE_PARAM);
+ const activeRunState = searchParams.get(DAG_RUN_STATE_PARAM);
const [pattern, setPattern] = useState("");
@@ -107,8 +109,8 @@ export const DagsFilters = () => {
setSearchParams(searchParams);
};
- const handleStateChange = (value: StateValue) => {
- if (value === "all") {
+ const handleStateChange = (value: string | undefined) => {
+ if (value === undefined) {
searchParams.delete(LAST_DAG_RUN_STATE_PARAM);
} else {
searchParams.set(LAST_DAG_RUN_STATE_PARAM, value);
@@ -117,6 +119,16 @@ export const DagsFilters = () => {
setSearchParams(searchParams);
};
+ const handleActiveRunChange = (value: string | undefined) => {
+ if (value === undefined) {
+ searchParams.delete(DAG_RUN_STATE_PARAM);
+ } else {
+ searchParams.set(DAG_RUN_STATE_PARAM, value);
+ }
+ resetPagination();
+ setSearchParams(searchParams);
+ };
+
const handleNeedsReviewToggle = () => {
if (needsReview === "true") {
searchParams.delete(NEEDS_REVIEW_PARAM);
@@ -140,15 +152,25 @@ export const DagsFilters = () => {
setTagFilterMode(checked ? "all" : "any");
};
- const stateValue = toStateValue(state);
const pausedValue = toBooleanFilterValue(showPaused, defaultShowPaused);
const favoriteValue = toBooleanFilterValue(showFavorites);
return (
<HStack flexWrap="wrap" gap={2} justifyContent="space-between">
- <Box overflowX="auto">
- <StateFilters onChange={handleStateChange} value={stateValue} />
- </Box>
+ <RunStateSelect
+ dataTestId="dags-last-run-state-filter"
+ label={translate("filters.lastRunState")}
+ onChange={handleStateChange}
+ states={lastRunStates}
+ value={state ?? undefined}
+ />
+ <RunStateSelect
+ dataTestId="dags-any-run-state-filter"
+ label={translate("filters.anyRunState")}
+ onChange={handleActiveRunChange}
+ states={anyRunStates}
+ value={activeRunState ?? undefined}
+ />
<RequiredActionFilter needsReview={needsReview === "true"}
onToggle={handleNeedsReviewToggle} />
<PausedFilter onChange={handlePausedChange} value={pausedValue} />
<TagFilter
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/RunStateSelect.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/RunStateSelect.tsx
new file mode 100644
index 00000000000..38a7cf59ede
--- /dev/null
+++
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/RunStateSelect.tsx
@@ -0,0 +1,87 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { HStack, Text } from "@chakra-ui/react";
+import { createListCollection } from "@chakra-ui/react/collection";
+import { useTranslation } from "react-i18next";
+import { FiList } from "react-icons/fi";
+
+import { StateBadge } from "src/components/StateBadge";
+import { Select } from "src/components/ui";
+
+type RunState = "failed" | "queued" | "running" | "success";
+
+type Props = {
+ readonly dataTestId?: string;
+ readonly label: string;
+ readonly onChange: (value: string | undefined) => void;
+ readonly states: ReadonlyArray<RunState>;
+ readonly value: string | undefined;
+};
+
+const ALL_VALUE = "all";
+
+export const RunStateSelect = ({ dataTestId, label, onChange, states, value }:
Props) => {
+ const { t: translate } = useTranslation(["dags", "common"]);
+
+ const collection = createListCollection({
+ items: [
+ { label: translate("dags:filters.allStates"), state: undefined, value:
ALL_VALUE },
+ ...states.map((state) => ({ label: translate(`common:states.${state}`),
state, value: state })),
+ ],
+ });
+
+ const current = collection.items.find((item) => item.value === (value ??
ALL_VALUE));
+
+ return (
+ <Select.Root
+ collection={collection}
+ data-testid={dataTestId}
+ minWidth="232px"
+ onValueChange={({ value: selected }) => onChange(selected[0] ===
ALL_VALUE ? undefined : selected[0])}
+ value={[value ?? ALL_VALUE]}
+ width="fit-content"
+ >
+ <Select.Trigger>
+ <HStack gap={2} justifyContent="space-between" pe={5} width="full">
+ <Text color="fg.muted" whiteSpace="nowrap">
+ {label}:
+ </Text>
+ <HStack gap={2}>
+ {current?.state === undefined ? <FiList /> : <StateBadge
state={current.state} />}
+ <Text whiteSpace="nowrap">{current?.label}</Text>
+ </HStack>
+ </HStack>
+ </Select.Trigger>
+ <Select.Content>
+ {collection.items.map((item) => (
+ <Select.Item
+ data-testid={dataTestId === undefined ? undefined :
`${dataTestId}-${item.value}`}
+ item={item}
+ key={item.value}
+ >
+ <HStack gap={2}>
+ {item.state === undefined ? <FiList /> : <StateBadge
state={item.state} />}
+ <Text>{item.label}</Text>
+ </HStack>
+ </Select.Item>
+ ))}
+ </Select.Content>
+ </Select.Root>
+ );
+};
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/StateFilters.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/StateFilters.tsx
deleted file mode 100644
index 47dc79953b8..00000000000
---
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/StateFilters.tsx
+++ /dev/null
@@ -1,75 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-import { useTranslation } from "react-i18next";
-
-import { StateBadge } from "src/components/StateBadge";
-import { ButtonGroupToggle, type ButtonGroupOption } from
"src/components/ui/ButtonGroupToggle";
-
-type StateValue = "all" | "failed" | "queued" | "running" | "success";
-
-type Props = {
- readonly onChange: (value: StateValue) => void;
- readonly value: StateValue;
-};
-
-export const StateFilters = ({ onChange, value }: Props) => {
- const { t: translate } = useTranslation(["dags", "common"]);
-
- const options: Array<ButtonGroupOption<StateValue>> = [
- { label: translate("dags:filters.paused.all"), value: "all" },
- {
- label: (
- <>
- <StateBadge state="failed" />
- {translate("common:states.failed")}
- </>
- ),
- value: "failed",
- },
- {
- label: (
- <>
- <StateBadge state="queued" />
- {translate("common:states.queued")}
- </>
- ),
- value: "queued",
- },
- {
- label: (
- <>
- <StateBadge state="running" />
- {translate("common:states.running")}
- </>
- ),
- value: "running",
- },
- {
- label: (
- <>
- <StateBadge state="success" />
- {translate("common:states.success")}
- </>
- ),
- value: "success",
- },
- ];
-
- return <ButtonGroupToggle<StateValue> onChange={onChange} options={options}
value={value} />;
-};
diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.test.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.test.tsx
index 1f840cc5d40..271542fc5c0 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.test.tsx
@@ -17,7 +17,7 @@
* under the License.
*/
import "@testing-library/jest-dom";
-import { render, screen, waitFor } from "@testing-library/react";
+import { render, screen, waitFor, within } from "@testing-library/react";
import { describe, it, expect } from "vitest";
import { AppWrapper } from "src/utils/AppWrapper";
@@ -26,12 +26,16 @@ describe("Dag Filters", () => {
it("Filter by selected last run state", async () => {
render(<AppWrapper initialEntries={["/dags"]} />);
- await waitFor(() =>
expect(screen.getByText("states.success")).toBeInTheDocument());
- await waitFor(() => screen.getByText("states.success").click());
await waitFor(() =>
expect(screen.getByText("tutorial_taskflow_api_success")).toBeInTheDocument());
- await waitFor(() =>
expect(screen.getByText("states.failed")).toBeInTheDocument());
- await waitFor(() => screen.getByText("states.failed").click());
+ const trigger =
within(screen.getByTestId("dags-last-run-state-filter")).getByRole("combobox");
+
+ await waitFor(() => trigger.click());
+ await waitFor(() =>
screen.getByTestId("dags-last-run-state-filter-success").click());
+ await waitFor(() =>
expect(screen.getByText("tutorial_taskflow_api_success")).toBeInTheDocument());
+
+ await waitFor(() => trigger.click());
+ await waitFor(() =>
screen.getByTestId("dags-last-run-state-filter-failed").click());
await waitFor(() =>
expect(screen.getByText("tutorial_taskflow_api_failed")).toBeInTheDocument());
});
});
diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
index 12e08fdb51b..8b315e99342 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
@@ -193,6 +193,7 @@ const createColumns = (
];
const {
+ DAG_RUN_STATE,
FAVORITE,
LAST_DAG_RUN_STATE,
NAME_PATTERN,
@@ -229,6 +230,7 @@ export const DagsList = () => {
const showFavorites = searchParams.get(FAVORITE);
const lastDagRunState = searchParams.get(LAST_DAG_RUN_STATE) as DagRunState;
+ const dagRunState = searchParams.get(DAG_RUN_STATE) as DagRunState;
const { selectedTags, tagFilterMode: selectedMatchMode } = useTagFilter();
const pendingReviews = searchParams.get(NEEDS_REVIEW);
const owners = searchParams.getAll(OWNERS);
@@ -284,6 +286,7 @@ export const DagsList = () => {
advancedSearch: advancedSearch.enabled,
dagDisplayNamePattern: Boolean(dagDisplayNamePattern) ?
dagDisplayNamePattern : undefined,
dagRunsLimit,
+ dagRunState,
isFavorite,
lastDagRunState,
limit: pagination.pageSize,
diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx
b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx
index 25545dfac50..cedbdb7ba64 100644
--- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/Stats.tsx
@@ -75,7 +75,7 @@ export const Stats = () => {
isLoading={isStatsLoading}
isRTL={isRTL}
label={translate("stats.queuedDags")}
- link="dags?last_dag_run_state=queued"
+ link="dags?dag_run_state=queued"
state="queued"
/>
) : undefined}
@@ -86,7 +86,7 @@ export const Stats = () => {
isLoading={isStatsLoading}
isRTL={isRTL}
label={translate("stats.runningDags")}
- link="dags?last_dag_run_state=running"
+ link="dags?dag_run_state=running"
state="running"
/>
diff --git a/airflow-core/src/airflow/ui/src/queries/useDags.tsx
b/airflow-core/src/airflow/ui/src/queries/useDags.tsx
index 4be8244ce73..ec4dfb18774 100644
--- a/airflow-core/src/airflow/ui/src/queries/useDags.tsx
+++ b/airflow-core/src/airflow/ui/src/queries/useDags.tsx
@@ -29,6 +29,7 @@ export const useDags = ({
dagDisplayNamePattern,
dagIdPattern,
dagRunsLimit,
+ dagRunState,
excludeStale = true,
isFavorite,
lastDagRunState,
@@ -45,6 +46,7 @@ export const useDags = ({
dagDisplayNamePattern?: string;
dagIdPattern?: string;
dagRunsLimit: number;
+ dagRunState?: DagRunState;
excludeStale?: boolean;
isFavorite?: boolean;
lastDagRunState?: DagRunState;
@@ -65,6 +67,7 @@ export const useDags = ({
? { dagDisplayNamePattern, dagIdPattern }
: { dagDisplayNamePrefixPattern: dagDisplayNamePattern,
dagIdPrefixPattern: dagIdPattern }),
dagRunsLimit,
+ dagRunState,
excludeStale,
hasPendingActions: pendingHitl,
isFavorite,
diff --git a/airflow-core/src/airflow/ui/tests/e2e/pages/DagsPage.ts
b/airflow-core/src/airflow/ui/tests/e2e/pages/DagsPage.ts
index f83134d6f93..bcf9d0114d3 100644
--- a/airflow-core/src/airflow/ui/tests/e2e/pages/DagsPage.ts
+++ b/airflow-core/src/airflow/ui/tests/e2e/pages/DagsPage.ts
@@ -32,17 +32,14 @@ export class DagsPage extends BasePage {
public readonly cardViewButton: Locator;
public readonly confirmButton: Locator;
- public readonly failedFilter: Locator;
public readonly hitlReviewModal: HITLReviewModal;
+ public readonly lastRunStateFilter: Locator;
public readonly needsReviewBadges: Locator;
public readonly needsReviewFilter: Locator;
public readonly operatorFilter: Locator;
- public readonly queuedFilter: Locator;
public readonly retriesFilter: Locator;
- public readonly runningFilter: Locator;
public readonly searchBox: Locator;
public readonly searchInput: Locator;
- public readonly successFilter: Locator;
public readonly tableViewButton: Locator;
public readonly triggerButton: Locator;
public readonly triggerRuleFilter: Locator;
@@ -64,12 +61,9 @@ export class DagsPage extends BasePage {
this.retriesFilter = page.getByTestId("retries-filter");
this.cardViewButton = page.getByRole("button", { name: "Show card view" });
this.tableViewButton = page.getByRole("button", { name: "Show table view"
});
- this.successFilter = page.getByRole("button", { name: "Success" });
- this.failedFilter = page.getByRole("button", { name: "Failed" });
+ this.lastRunStateFilter = page.getByTestId("dags-last-run-state-filter");
this.hitlReviewModal = new HITLReviewModal(page);
this.needsReviewBadges = page.getByTestId("needs-review-badge");
- this.runningFilter = page.getByRole("button", { name: "Running" });
- this.queuedFilter = page.getByRole("button", { name: "Queued" });
// Uses testId because this button's text is driven by an i18n key.
this.needsReviewFilter = page.getByTestId("dags-needs-review-filter");
}
@@ -118,14 +112,6 @@ export class DagsPage extends BasePage {
public async filterByStatus(
status: "failed" | "needs_review" | "queued" | "running" | "success",
): Promise<void> {
- const filterMap: Record<typeof status, Locator> = {
- failed: this.failedFilter,
- needs_review: this.needsReviewFilter,
- queued: this.queuedFilter,
- running: this.runningFilter,
- success: this.successFilter,
- };
-
// Set up response listener before the click so we don't miss a fast
response.
const responsePromise = this.page
.waitForResponse((resp: Response) => resp.url().includes("/api/v2/dags")
&& resp.status() === 200, {
@@ -137,7 +123,13 @@ export class DagsPage extends BasePage {
}
});
- await filterMap[status].click();
+ if (status === "needs_review") {
+ await this.needsReviewFilter.click();
+ } else {
+ // Run-state filters live in the "Last run" dropdown.
+ await this.lastRunStateFilter.getByRole("combobox").click();
+ await
this.page.getByTestId(`dags-last-run-state-filter-${status}`).click();
+ }
await responsePromise;
}
diff --git a/airflow-core/src/airflow/ui/tests/e2e/specs/dags-list.spec.ts
b/airflow-core/src/airflow/ui/tests/e2e/specs/dags-list.spec.ts
index 798a910a52e..5198bb6e26a 100644
--- a/airflow-core/src/airflow/ui/tests/e2e/specs/dags-list.spec.ts
+++ b/airflow-core/src/airflow/ui/tests/e2e/specs/dags-list.spec.ts
@@ -192,15 +192,12 @@ test.describe("Dags Search", () => {
});
test.describe("Dags Status Filtering", () => {
- test("should display status filter buttons", async ({ dagsPage }) => {
+ test("should filter Dags by run state", async ({ dagsPage }) => {
test.slow();
await dagsPage.navigate();
await dagsPage.waitForDagList();
- await expect(dagsPage.successFilter).toBeVisible();
- await expect(dagsPage.failedFilter).toBeVisible();
- await expect(dagsPage.runningFilter).toBeVisible();
- await expect(dagsPage.queuedFilter).toBeVisible();
+ await expect(dagsPage.lastRunStateFilter).toBeVisible();
await dagsPage.filterByStatus("success");
await dagsPage.waitForDagList();
diff --git a/airflow-core/src/airflow/ui/tests/e2e/specs/home-dashboard.spec.ts
b/airflow-core/src/airflow/ui/tests/e2e/specs/home-dashboard.spec.ts
index e74669e6b77..01169854e90 100644
--- a/airflow-core/src/airflow/ui/tests/e2e/specs/home-dashboard.spec.ts
+++ b/airflow-core/src/airflow/ui/tests/e2e/specs/home-dashboard.spec.ts
@@ -64,7 +64,7 @@ test.describe("Dashboard Metrics Display", () => {
await homePage.waitForDashboardLoad();
await homePage.runningDagsCard.click();
- await expect(homePage.page).toHaveURL(/last_dag_run_state=running/);
+ await expect(homePage.page).toHaveURL(/dag_run_state=running/);
});
test("should display welcome heading on dashboard", async ({ homePage }) => {
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 50444e04914..4e561fd3436 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
@@ -22,6 +22,7 @@ from unittest import mock
import pendulum
import pytest
from fastapi.testclient import TestClient
+from sqlalchemy import select
from sqlalchemy.orm import Session
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
@@ -134,6 +135,31 @@ class TestGetDagRuns(TestPublicDagEndpoint):
assert previous_run_after > dag_run["run_after"]
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.
+ older_run = session.scalar(
+ select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
"run_id_1")
+ )
+ older_run.state = DagRunState.RUNNING
+ 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"})
+ 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"})
+ 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."""
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dashboard.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dashboard.py
index da46c51da1b..50869e0ee9f 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dashboard.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dashboard.py
@@ -381,7 +381,7 @@ class TestDagStatsEndpoint:
assert response.json() == {
"active_dag_count": 1,
"failed_dag_count": 0,
- "running_dag_count": 0,
+ "running_dag_count": 1,
"queued_dag_count": 1,
}
