This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 4f4befcc73c Show per-Dag run state counts on the Dags list page
(#67242)
4f4befcc73c is described below
commit 4f4befcc73cac7786424867eb7fa97a6f75352c4
Author: Yuseok Jo <[email protected]>
AuthorDate: Tue Jul 7 07:35:13 2026 +0900
Show per-Dag run state counts on the Dags list page (#67242)
* UI: Show per-Dag run state counts on the Dags list page
* UI: Align run history chart to card bottom in Dag list
* UI: Add time range filter for run state counts on Dags list
* UI: Fix openapi-gen queries runAfterGte type and formatting for run state
counts
* UI: Fix Dag capitalization and remove stale performance comment in run
state counts
* UI: Simplify Dag ID deduplication with set intersection in run state
counts
* Cap per-Dag run state counts to scale on large DagRun tables
* Show capped per-Dag run state counts as "N+" on the Dags list
* Query run state counts per state to keep them fast at scale
* Apply prettier formatting to run state counts UI components
* Drop feature newsfragment for Dags list run state counts
* Drop redundant useMemo for dag ids in run state counts query
* UI: Drop redundant useMemo in Dags list run state counts (React Compiler
memoizes)
* Regenerate run state counts API client after rebase on main
* Remove time range filter from Dags list run state counts
* Cap run_state_counts dag_ids to the maximum page limit
* Revert unused showDateRange prop from shared TimeRangeSelector
---
.../api_fastapi/core_api/datamodels/ui/dags.py | 15 +++
.../api_fastapi/core_api/openapi/_private_ui.yaml | 69 ++++++++++
.../airflow/api_fastapi/core_api/routes/ui/dags.py | 62 ++++++++-
.../src/airflow/ui/openapi-gen/queries/common.ts | 6 +
.../ui/openapi-gen/queries/ensureQueryData.ts | 11 ++
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 11 ++
.../src/airflow/ui/openapi-gen/queries/queries.ts | 11 ++
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 11 ++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 43 +++++++
.../ui/openapi-gen/requests/services.gen.ts | 23 +++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 39 ++++++
.../airflow/ui/public/i18n/locales/en/dags.json | 5 +
.../airflow/ui/src/pages/DagsList/DagCard.test.tsx | 27 ++--
.../src/airflow/ui/src/pages/DagsList/DagCard.tsx | 98 ++++++++------
.../src/pages/DagsList/DagRunStateCounts.test.tsx | 100 ++++++++++++++
.../ui/src/pages/DagsList/DagRunStateCounts.tsx | 87 +++++++++++++
.../src/airflow/ui/src/pages/DagsList/DagsList.tsx | 63 +++++++--
.../ui/src/queries/useDagRunStateCounts.tsx | 45 +++++++
.../api_fastapi/core_api/routes/ui/test_dags.py | 143 +++++++++++++++++++++
19 files changed, 811 insertions(+), 58 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py
index febdf45c0a8..590084cc2c5 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py
@@ -22,6 +22,7 @@ from airflow.api_fastapi.core_api.datamodels.common import
MaybeAssetExpression
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail
from airflow.api_fastapi.core_api.datamodels.ui.dag_runs import
DAGRunLightResponse
+from airflow.utils.state import DagRunState
class DAGWithLatestDagRunsResponse(DAGResponse):
@@ -38,3 +39,17 @@ class DAGWithLatestDagRunsCollectionResponse(BaseModel):
total_entries: int
dags: list[DAGWithLatestDagRunsResponse]
+
+
+class DAGRunStateCountsResponse(BaseModel):
+ """Per-Dag counts of DagRuns grouped by state."""
+
+ dag_id: str
+ state_counts: dict[DagRunState, int]
+
+
+class DAGsRunStateCountsCollectionResponse(BaseModel):
+ """Collection of per-Dag DagRun-state counts for the Dag list page."""
+
+ dags: list[DAGRunStateCountsResponse]
+ state_count_limit: int
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 45da42f9224..887aee12c9b 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
@@ -593,6 +593,41 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ /ui/dags/run_state_counts:
+ get:
+ tags:
+ - DAG
+ summary: Get Dag Run State Counts
+ description: Return per-Dag DagRun state counts (zero-filled) for the
Dag list
+ page.
+ operationId: get_dag_run_state_counts_ui
+ security:
+ - OAuth2PasswordBearer: []
+ - HTTPBearer: []
+ parameters:
+ - name: dag_ids
+ in: query
+ required: true
+ schema:
+ type: array
+ items:
+ type: string
+ minItems: 1
+ maxItems: 100
+ title: Dag Ids
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref:
'#/components/schemas/DAGsRunStateCountsCollectionResponse'
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/ui/dependencies:
get:
tags:
@@ -2260,6 +2295,24 @@ components:
- duration
title: DAGRunLightResponse
description: DAG Run serializer for responses.
+ DAGRunStateCountsResponse:
+ properties:
+ dag_id:
+ type: string
+ title: Dag Id
+ state_counts:
+ additionalProperties:
+ type: integer
+ propertyNames:
+ $ref: '#/components/schemas/DagRunState'
+ type: object
+ title: State Counts
+ type: object
+ required:
+ - dag_id
+ - state_counts
+ title: DAGRunStateCountsResponse
+ description: Per-Dag counts of DagRuns grouped by state.
DAGRunStates:
properties:
queued:
@@ -2499,6 +2552,22 @@ components:
- file_token
title: DAGWithLatestDagRunsResponse
description: DAG with latest dag runs response serializer.
+ DAGsRunStateCountsCollectionResponse:
+ properties:
+ dags:
+ items:
+ $ref: '#/components/schemas/DAGRunStateCountsResponse'
+ type: array
+ title: Dags
+ state_count_limit:
+ type: integer
+ title: State Count Limit
+ type: object
+ required:
+ - dags
+ - state_count_limit
+ title: DAGsRunStateCountsCollectionResponse
+ description: Collection of per-Dag DagRun-state counts for the Dag list
page.
DagRunState:
type: string
enum:
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 d2b86beb2b7..15515a92bb8 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
@@ -19,8 +19,8 @@ from __future__ import annotations
from typing import Annotated
-from fastapi import Depends, HTTPException, status
-from sqlalchemy import select, union_all
+from fastapi import Depends, HTTPException, Query, status
+from sqlalchemy import func, literal, select, union_all
from sqlalchemy.orm import defaultload
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
@@ -57,20 +57,24 @@ from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.datamodels.dags import DAG_ALIAS_MAPPING,
DAGResponse
from airflow.api_fastapi.core_api.datamodels.ui.dag_runs import
DAGRunLightResponse
from airflow.api_fastapi.core_api.datamodels.ui.dags import (
+ DAGRunStateCountsResponse,
+ DAGsRunStateCountsCollectionResponse,
DAGWithLatestDagRunsCollectionResponse,
DAGWithLatestDagRunsResponse,
)
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.routes.ui.dashboard import STATE_COUNT_CAP
from airflow.api_fastapi.core_api.security import (
GetUserDep,
ReadableDagsFilterDep,
requires_access_dag,
)
+from airflow.configuration import conf
from airflow.models import DagModel, DagRun
from airflow.models.dag_favorite import DagFavorite
from airflow.models.hitl import HITLDetail
from airflow.models.taskinstance import TaskInstance
-from airflow.utils.state import TaskInstanceState
+from airflow.utils.state import DagRunState, TaskInstanceState
dags_router = AirflowRouter(prefix="/dags", tags=["DAG"])
@@ -284,3 +288,55 @@ def get_latest_run_info(dag_id: str, session: SessionDep)
-> DAGRunLightResponse
latest_run_info = session.execute(latest_run_info_select).one_or_none()
return DAGRunLightResponse(**latest_run_info._mapping) if latest_run_info
else None
+
+
+@dags_router.get(
+ "/run_state_counts",
+ dependencies=[
+ Depends(requires_access_dag(method="GET")),
+ Depends(requires_access_dag(method="GET",
access_entity=DagAccessEntity.RUN)),
+ ],
+ operation_id="get_dag_run_state_counts_ui",
+)
+def get_dag_run_state_counts(
+ session: SessionDep,
+ readable_dags_filter: ReadableDagsFilterDep,
+ dag_ids: Annotated[list[str], Query(min_length=1,
max_length=conf.getint("api", "maximum_page_limit"))],
+) -> DAGsRunStateCountsCollectionResponse:
+ """Return per-Dag DagRun state counts (zero-filled) for the Dag list
page."""
+ permitted_dag_ids = readable_dags_filter.value or set()
+ requested_dag_ids = list(set(dag_ids) & permitted_dag_ids)
+ counts_by_dag: dict[str, dict[DagRunState, int]] = {
+ dag_id: {state: 0 for state in DagRunState} for dag_id in
requested_dag_ids
+ }
+
+ if requested_dag_ids:
+ # One capped union per state, not a single (dag, state) union: keeping
every
+ # branch on the same state lets the planner pick a uniform, efficient
per-branch
+ # plan. A mixed-state union misplans and scans whole partitions
(orders of
+ # magnitude slower). Each branch reads at most STATE_COUNT_CAP rows;
the UI
+ # shows "N+" once a count reaches the cap.
+ for state in DagRunState:
+ branches = []
+ for dag_id in requested_dag_ids:
+ capped = (
+ select(literal(dag_id).label("dag_id"))
+ .select_from(DagRun)
+ .where(DagRun.dag_id == dag_id, DagRun.state == state)
+ .limit(STATE_COUNT_CAP)
+ .subquery()
+ )
+ branches.append(select(capped.c.dag_id))
+ counts = union_all(*branches).subquery()
+ for row in session.execute(
+ select(counts.c.dag_id,
func.count().label("cnt")).group_by(counts.c.dag_id)
+ ):
+ counts_by_dag[row.dag_id][state] = row.cnt
+
+ return DAGsRunStateCountsCollectionResponse(
+ dags=[
+ DAGRunStateCountsResponse(dag_id=dag_id, state_counts=counts)
+ for dag_id, counts in counts_by_dag.items()
+ ],
+ state_count_limit=STATE_COUNT_CAP,
+ )
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 69372a087a3..a34ca6b0c68 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -356,6 +356,12 @@ export const useDagServiceGetLatestRunInfoKey =
"DagServiceGetLatestRunInfo";
export const UseDagServiceGetLatestRunInfoKeyFn = ({ dagId }: {
dagId: string;
}, queryKey?: Array<unknown>) => [useDagServiceGetLatestRunInfoKey,
...(queryKey ?? [{ dagId }])];
+export type DagServiceGetDagRunStateCountsUiDefaultResponse =
Awaited<ReturnType<typeof DagService.getDagRunStateCountsUi>>;
+export type DagServiceGetDagRunStateCountsUiQueryResult<TData =
DagServiceGetDagRunStateCountsUiDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
+export const useDagServiceGetDagRunStateCountsUiKey =
"DagServiceGetDagRunStateCountsUi";
+export const UseDagServiceGetDagRunStateCountsUiKeyFn = ({ dagIds }: {
+ dagIds: string[];
+}, queryKey?: Array<unknown>) => [useDagServiceGetDagRunStateCountsUiKey,
...(queryKey ?? [{ dagIds }])];
export type EventLogServiceGetEventLogDefaultResponse =
Awaited<ReturnType<typeof EventLogService.getEventLog>>;
export type EventLogServiceGetEventLogQueryResult<TData =
EventLogServiceGetEventLogDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useEventLogServiceGetEventLogKey = "EventLogServiceGetEventLog";
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 ff0174b77e9..c4c043c93b4 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -709,6 +709,17 @@ export const ensureUseDagServiceGetLatestRunInfoData =
(queryClient: QueryClient
dagId: string;
}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagServiceGetLatestRunInfoKeyFn({ dagId }), queryFn: () =>
DagService.getLatestRunInfo({ dagId }) });
/**
+* Get Dag Run State Counts
+* Return per-Dag DagRun state counts (zero-filled) for the Dag list page.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsRunStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseDagServiceGetDagRunStateCountsUiData = (queryClient:
QueryClient, { dagIds }: {
+ dagIds: string[];
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () =>
DagService.getDagRunStateCountsUi({ dagIds }) });
+/**
* Get Event Log
* @param data The data for the request.
* @param data.eventLogId
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 1f8db278d9d..7aebe21755c 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -709,6 +709,17 @@ export const prefetchUseDagServiceGetLatestRunInfo =
(queryClient: QueryClient,
dagId: string;
}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagServiceGetLatestRunInfoKeyFn({ dagId }), queryFn: () =>
DagService.getLatestRunInfo({ dagId }) });
/**
+* Get Dag Run State Counts
+* Return per-Dag DagRun state counts (zero-filled) for the Dag list page.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsRunStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseDagServiceGetDagRunStateCountsUi = (queryClient:
QueryClient, { dagIds }: {
+ dagIds: string[];
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () =>
DagService.getDagRunStateCountsUi({ dagIds }) });
+/**
* Get Event Log
* @param data The data for the request.
* @param data.eventLogId
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 c5ec74825be..42eed251fa4 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -709,6 +709,17 @@ export const useDagServiceGetLatestRunInfo = <TData =
Common.DagServiceGetLatest
dagId: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetLatestRunInfoKeyFn({ dagId }, queryKey), queryFn: () =>
DagService.getLatestRunInfo({ dagId }) as TData, ...options });
/**
+* Get Dag Run State Counts
+* Return per-Dag DagRun state counts (zero-filled) for the Dag list page.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsRunStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetDagRunStateCountsUi = <TData =
Common.DagServiceGetDagRunStateCountsUiDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagIds }: {
+ dagIds: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }, queryKey), queryFn:
() => DagService.getDagRunStateCountsUi({ dagIds }) as TData, ...options });
+/**
* Get Event Log
* @param data The data for the request.
* @param data.eventLogId
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 90f75e1cfb3..b3f2d6f4e9f 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -709,6 +709,17 @@ export const useDagServiceGetLatestRunInfoSuspense =
<TData = Common.DagServiceG
dagId: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetLatestRunInfoKeyFn({ dagId }, queryKey), queryFn: () =>
DagService.getLatestRunInfo({ dagId }) as TData, ...options });
/**
+* Get Dag Run State Counts
+* Return per-Dag DagRun state counts (zero-filled) for the Dag list page.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsRunStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetDagRunStateCountsUiSuspense = <TData =
Common.DagServiceGetDagRunStateCountsUiDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagIds }: {
+ dagIds: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }, queryKey), queryFn:
() => DagService.getDagRunStateCountsUi({ dagIds }) as TData, ...options });
+/**
* Get Event Log
* @param data The data for the request.
* @param data.eventLogId
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 08dbfa80a44..4dab4ff145c 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
@@ -8898,6 +8898,29 @@ export const $DAGRunLightResponse = {
description: 'DAG Run serializer for responses.'
} as const;
+export const $DAGRunStateCountsResponse = {
+ properties: {
+ dag_id: {
+ type: 'string',
+ title: 'Dag Id'
+ },
+ state_counts: {
+ additionalProperties: {
+ type: 'integer'
+ },
+ propertyNames: {
+ '$ref': '#/components/schemas/DagRunState'
+ },
+ type: 'object',
+ title: 'State Counts'
+ }
+ },
+ type: 'object',
+ required: ['dag_id', 'state_counts'],
+ title: 'DAGRunStateCountsResponse',
+ description: 'Per-Dag counts of DagRuns grouped by state.'
+} as const;
+
export const $DAGRunStates = {
properties: {
queued: {
@@ -9248,6 +9271,26 @@ export const $DAGWithLatestDagRunsResponse = {
description: 'DAG with latest dag runs response serializer.'
} as const;
+export const $DAGsRunStateCountsCollectionResponse = {
+ properties: {
+ dags: {
+ items: {
+ '$ref': '#/components/schemas/DAGRunStateCountsResponse'
+ },
+ type: 'array',
+ title: 'Dags'
+ },
+ state_count_limit: {
+ type: 'integer',
+ title: 'State Count Limit'
+ }
+ },
+ type: 'object',
+ required: ['dags', 'state_count_limit'],
+ title: 'DAGsRunStateCountsCollectionResponse',
+ description: 'Collection of per-Dag DagRun-state counts for the Dag list
page.'
+} as const;
+
export const $DagRunStatsResponse = {
properties: {
duration: {
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 ebb30ffa68c..b44b9865237 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
@@ -3,7 +3,7 @@
import type { CancelablePromise } from './core/CancelablePromise';
import { OpenAPI } from './core/OpenAPI';
import { request as __request } from './core/request';
-import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
+import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
export class AssetService {
/**
@@ -1987,6 +1987,27 @@ export class DagService {
});
}
+ /**
+ * Get Dag Run State Counts
+ * Return per-Dag DagRun state counts (zero-filled) for the Dag list page.
+ * @param data The data for the request.
+ * @param data.dagIds
+ * @returns DAGsRunStateCountsCollectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static getDagRunStateCountsUi(data: GetDagRunStateCountsUiData):
CancelablePromise<GetDagRunStateCountsUiResponse> {
+ return __request(OpenAPI, {
+ method: 'GET',
+ url: '/ui/dags/run_state_counts',
+ query: {
+ dag_ids: data.dagIds
+ },
+ errors: {
+ 422: 'Validation Error'
+ }
+ });
+ }
+
}
export class EventLogService {
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 a9943390b27..a9d002e1e3f 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
@@ -2271,6 +2271,16 @@ export type DAGRunLightResponse = {
readonly duration: number | null;
};
+/**
+ * Per-Dag counts of DagRuns grouped by state.
+ */
+export type DAGRunStateCountsResponse = {
+ dag_id: string;
+ state_counts: {
+ [key: string]: (number);
+ };
+};
+
/**
* DAG Run States for responses.
*/
@@ -2335,6 +2345,14 @@ export type DAGWithLatestDagRunsResponse = {
readonly file_token: string;
};
+/**
+ * Collection of per-Dag DagRun-state counts for the Dag list page.
+ */
+export type DAGsRunStateCountsCollectionResponse = {
+ dags: Array<DAGRunStateCountsResponse>;
+ state_count_limit: number;
+};
+
/**
* DAG Run statistics serializer for responses.
*/
@@ -3483,6 +3501,12 @@ export type GetLatestRunInfoData = {
export type GetLatestRunInfoResponse = DAGRunLightResponse | null;
+export type GetDagRunStateCountsUiData = {
+ dagIds: Array<(string)>;
+};
+
+export type GetDagRunStateCountsUiResponse =
DAGsRunStateCountsCollectionResponse;
+
export type GetEventLogData = {
eventLogId: number;
};
@@ -6341,6 +6365,21 @@ export type $OpenApiTs = {
};
};
};
+ '/ui/dags/run_state_counts': {
+ get: {
+ req: GetDagRunStateCountsUiData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: DAGsRunStateCountsCollectionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
'/api/v2/eventLogs/{event_log_id}': {
get: {
req: GetEventLogData;
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 30fefed93f5..990d6c7848f 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
@@ -76,6 +76,11 @@
"upstream": "Upstream"
}
},
+ "runStateCounts": {
+ "label": "Run states",
+ "loading": "Loading run state counts",
+ "tooltip": "{{state}}: {{formattedCount}} — click to filter runs"
+ },
"search": {
"advanced": "Advanced Search",
"clear": "Clear search",
diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.test.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.test.tsx
index 14532199258..ad163ae19f1 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.test.tsx
@@ -53,6 +53,15 @@ const GMTWrapper = ({ children }: PropsWithChildren) => (
</BaseWrapper>
);
+// Render with the run-state-counts row in its loading state so it renders
+// skeletons rather than StateBadges. Without this, every card would emit
+// 4 extra "state-badge" testids and break getByTestId assertions in tests
+// that target the latest-run badge.
+const renderCard = (dag: DAGWithLatestDagRunsResponse) =>
+ render(<DagCard dag={dag} runStateCounts={undefined} runStateCountsLoading
stateCountLimit={undefined} />, {
+ wrapper: GMTWrapper,
+ });
+
const mockDag = {
allowed_run_types: null,
asset_expression: null,
@@ -136,7 +145,7 @@ afterEach(() => {
describe("DagCard", () => {
it("DagCard should render without tags", () => {
- render(<DagCard dag={mockDag} />, { wrapper: GMTWrapper });
+ renderCard(mockDag);
expect(screen.getByText(mockDag.dag_display_name)).toBeInTheDocument();
expect(screen.queryByTestId("dag-tag")).toBeNull();
});
@@ -154,7 +163,7 @@ describe("DagCard", () => {
tags,
} satisfies DAGWithLatestDagRunsResponse;
- render(<DagCard dag={expandedMockDag} />, { wrapper: GMTWrapper });
+ renderCard(expandedMockDag);
expect(screen.getByTestId("dag-id")).toBeInTheDocument();
expect(screen.getByTestId("dag-tag")).toBeInTheDocument();
expect(screen.queryByText("tag3")).toBeInTheDocument();
@@ -176,14 +185,14 @@ describe("DagCard", () => {
tags,
} satisfies DAGWithLatestDagRunsResponse;
- render(<DagCard dag={expandedMockDag} />, { wrapper: GMTWrapper });
+ renderCard(expandedMockDag);
expect(screen.getByTestId("dag-id")).toBeInTheDocument();
expect(screen.getByTestId("dag-tag")).toBeInTheDocument();
expect(screen.getByText("+2 more")).toBeInTheDocument();
});
it("DagCard should render schedule section", () => {
- render(<DagCard dag={mockDag} />, { wrapper: GMTWrapper });
+ renderCard(mockDag);
const scheduleElement = screen.getByTestId("schedule");
expect(scheduleElement).toBeInTheDocument();
@@ -192,7 +201,7 @@ describe("DagCard", () => {
});
it("DagCard should render latest run section with actual run data", () => {
- render(<DagCard dag={mockDag} />, { wrapper: GMTWrapper });
+ renderCard(mockDag);
const latestRunElement = screen.getByTestId("latest-run");
expect(latestRunElement).toBeInTheDocument();
@@ -201,7 +210,7 @@ describe("DagCard", () => {
});
it("DagCard should render next run section with timestamp", () => {
- render(<DagCard dag={mockDag} />, { wrapper: GMTWrapper });
+ renderCard(mockDag);
const nextRunElement = screen.getByTestId("next-run");
expect(nextRunElement).toBeInTheDocument();
@@ -210,7 +219,7 @@ describe("DagCard", () => {
});
it("DagCard should not render next run timestamp for a paused Dag", () => {
- render(<DagCard dag={{ ...mockDag, is_paused: true }} />, { wrapper:
GMTWrapper });
+ renderCard({ ...mockDag, is_paused: true });
const nextRunElement = screen.getByTestId("next-run");
expect(nextRunElement).toBeInTheDocument();
@@ -218,7 +227,7 @@ describe("DagCard", () => {
});
it("DagCard should render StateBadge as success", () => {
- render(<DagCard dag={mockDag} />, { wrapper: GMTWrapper });
+ renderCard(mockDag);
const stateBadge = screen.getByTestId("state-badge");
expect(stateBadge).toBeInTheDocument();
@@ -243,7 +252,7 @@ describe("DagCard", () => {
],
} satisfies DAGWithLatestDagRunsResponse;
- render(<DagCard dag={mockDagWithFailedRun} />, { wrapper: GMTWrapper });
+ renderCard(mockDagWithFailedRun);
const stateBadge = screen.getByTestId("state-badge");
expect(stateBadge).toBeInTheDocument();
diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.tsx
index 9eef74269d6..cd320d417d0 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagCard.tsx
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { Box, Flex, HStack, SimpleGrid, Spinner } from "@chakra-ui/react";
+import { Box, Flex, Grid, GridItem, HStack, Spinner } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
import type { DAGWithLatestDagRunsResponse } from "openapi/requests/types.gen";
@@ -30,15 +30,19 @@ import { TriggerDAGButton } from
"src/components/TriggerDag/TriggerDAGButton";
import { RouterLink, Tooltip } from "src/components/ui";
import { isStatePending, useAutoRefresh } from "src/utils";
+import { DagRunStateCounts } from "./DagRunStateCounts";
import { DagTags } from "./DagTags";
import { RecentRuns } from "./RecentRuns";
import { Schedule } from "./Schedule";
type Props = {
readonly dag: DAGWithLatestDagRunsResponse;
+ readonly runStateCounts: Record<string, number> | undefined;
+ readonly runStateCountsLoading: boolean;
+ readonly stateCountLimit: number | undefined;
};
-export const DagCard = ({ dag }: Props) => {
+export const DagCard = ({ dag, runStateCounts, runStateCountsLoading,
stateCountLimit }: Props) => {
const { t: translate } = useTranslation(["common", "dag"]);
const [latestRun] = dag.latest_dag_runs;
@@ -74,42 +78,64 @@ export const DagCard = ({ dag }: Props) => {
<DeleteDagButton dagDisplayName={dag.dag_display_name}
dagId={dag.dag_id} />
</HStack>
</Flex>
- <SimpleGrid columns={4} gap={1} height={20} px={3} py={1}>
- <Stat data-testid="schedule" label={translate("dagDetails.schedule")}>
- <Schedule
- assetExpression={dag.asset_expression}
- dagId={dag.dag_id}
- timetableDescription={dag.timetable_description}
- timetablePartitioned={dag.timetable_partitioned}
- timetableSummary={dag.timetable_summary}
- />
- </Stat>
- <Stat data-testid="latest-run"
label={translate("dagDetails.latestRun")}>
- {latestRun ? (
- <RouterLink
to={`/dags/${latestRun.dag_id}/runs/${latestRun.run_id}`}>
+ <Grid gap={1} px={3} py={2} templateColumns="repeat(4, 1fr)"
templateRows="auto auto">
+ <GridItem gridColumn={1} gridRow={1}>
+ <Stat data-testid="schedule"
label={translate("dagDetails.schedule")}>
+ <Schedule
+ assetExpression={dag.asset_expression}
+ dagId={dag.dag_id}
+ timetableDescription={dag.timetable_description}
+ timetablePartitioned={dag.timetable_partitioned}
+ timetableSummary={dag.timetable_summary}
+ />
+ </Stat>
+ </GridItem>
+ <GridItem gridColumn={2} gridRow={1}>
+ <Stat data-testid="latest-run"
label={translate("dagDetails.latestRun")}>
+ {latestRun ? (
+ <RouterLink
to={`/dags/${latestRun.dag_id}/runs/${latestRun.run_id}`}>
+ <DagRunInfo
+ endDate={latestRun.end_date}
+ logicalDate={latestRun.logical_date}
+ runAfter={latestRun.run_after}
+ startDate={latestRun.start_date}
+ state={latestRun.state}
+ />
+ {isStatePending(latestRun.state) && !dag.is_paused &&
Boolean(refetchInterval) ? (
+ <Spinner />
+ ) : undefined}
+ </RouterLink>
+ ) : undefined}
+ </Stat>
+ </GridItem>
+ <GridItem gridColumn={3} gridRow={1}>
+ <Stat data-testid="next-run" label={translate("dagDetails.nextRun")}>
+ {!dag.is_paused && Boolean(dag.next_dagrun_run_after) ? (
<DagRunInfo
- endDate={latestRun.end_date}
- logicalDate={latestRun.logical_date}
- runAfter={latestRun.run_after}
- startDate={latestRun.start_date}
- state={latestRun.state}
+ logicalDate={dag.next_dagrun_logical_date}
+ runAfter={dag.next_dagrun_run_after as string}
/>
- {isStatePending(latestRun.state) && !dag.is_paused &&
Boolean(refetchInterval) ? (
- <Spinner />
- ) : undefined}
- </RouterLink>
- ) : undefined}
- </Stat>
- <Stat data-testid="next-run" label={translate("dagDetails.nextRun")}>
- {!dag.is_paused && Boolean(dag.next_dagrun_run_after) ? (
- <DagRunInfo
- logicalDate={dag.next_dagrun_logical_date}
- runAfter={dag.next_dagrun_run_after as string}
- />
- ) : undefined}
- </Stat>
- <RecentRuns latestRuns={dag.latest_dag_runs} />
- </SimpleGrid>
+ ) : undefined}
+ </Stat>
+ </GridItem>
+ <GridItem
+ alignItems="flex-end"
+ display="flex"
+ gridColumn={4}
+ gridRow="1 / 3"
+ justifyContent="flex-end"
+ >
+ <RecentRuns latestRuns={dag.latest_dag_runs} />
+ </GridItem>
+ <GridItem alignSelf="end" gridColumn={1} gridRow={2}>
+ <DagRunStateCounts
+ counts={runStateCounts}
+ dagId={dag.dag_id}
+ isLoading={runStateCountsLoading}
+ stateCountLimit={stateCountLimit}
+ />
+ </GridItem>
+ </Grid>
</Box>
);
};
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.test.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.test.tsx
new file mode 100644
index 00000000000..906b0b80937
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.test.tsx
@@ -0,0 +1,100 @@
+/*!
+ * 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 "@testing-library/jest-dom/vitest";
+import { render, screen } from "@testing-library/react";
+import { MemoryRouter } from "react-router-dom";
+import { describe, expect, it } from "vitest";
+
+import { BaseWrapper } from "src/utils/Wrapper";
+
+import "../../i18n/config";
+import { DagRunStateCounts } from "./DagRunStateCounts";
+
+const renderCounts = (
+ counts: Record<string, number> | undefined,
+ options: { isLoading?: boolean; stateCountLimit?: number } = {},
+) =>
+ render(
+ <DagRunStateCounts
+ counts={counts}
+ dagId="my_dag"
+ isLoading={options.isLoading ?? false}
+ stateCountLimit={options.stateCountLimit}
+ />,
+ {
+ wrapper: ({ children }) => (
+ <BaseWrapper>
+ <MemoryRouter>{children}</MemoryRouter>
+ </BaseWrapper>
+ ),
+ },
+ );
+
+describe("DagRunStateCounts", () => {
+ it("renders skeleton placeholders while loading", () => {
+ renderCounts(undefined, { isLoading: true });
+
expect(screen.getByTestId("run-state-counts-loading-my_dag")).toBeInTheDocument();
+ expect(screen.queryByTestId("run-state-counts-my_dag")).toBeNull();
+ });
+
+ it("renders one clickable badge per state with counts", () => {
+ renderCounts({ failed: 3, queued: 0, running: 1, success: 42 });
+ expect(screen.getByTestId("run-state-counts-my_dag")).toBeInTheDocument();
+
+ const successLink = screen.getByTestId("run-state-count-success-my_dag");
+ const failedLink = screen.getByTestId("run-state-count-failed-my_dag");
+ const runningLink = screen.getByTestId("run-state-count-running-my_dag");
+ const queuedLink = screen.getByTestId("run-state-count-queued-my_dag");
+
+ expect(successLink).toHaveAttribute("href",
"/dags/my_dag/runs?state=success");
+ expect(failedLink).toHaveAttribute("href",
"/dags/my_dag/runs?state=failed");
+ expect(runningLink).toHaveAttribute("href",
"/dags/my_dag/runs?state=running");
+ expect(queuedLink).toHaveAttribute("href",
"/dags/my_dag/runs?state=queued");
+
+ expect(successLink).toHaveTextContent("42");
+ expect(failedLink).toHaveTextContent("3");
+ expect(runningLink).toHaveTextContent("1");
+ expect(queuedLink).toHaveTextContent("0");
+ });
+
+ it("dims zero-count badges but keeps them clickable", () => {
+ renderCounts({ failed: 0, queued: 0, running: 0, success: 5 });
+ const queuedLink = screen.getByTestId("run-state-count-queued-my_dag");
+ // The badge is the link's child Chakra Badge; check style is dimmed.
+ const badge = queuedLink.querySelector('[data-testid="state-badge"]');
+
+ expect(badge).not.toBeNull();
+ expect(badge).toHaveStyle({ opacity: "0.4" });
+ // Still a real link so operators can confirm "no failures" by clicking
through.
+ expect(queuedLink).toHaveAttribute("href",
"/dags/my_dag/runs?state=queued");
+ });
+
+ it("treats missing keys in `counts` as zero", () => {
+ renderCounts({ failed: 2 });
+
expect(screen.getByTestId("run-state-count-success-my_dag")).toHaveTextContent("0");
+
expect(screen.getByTestId("run-state-count-failed-my_dag")).toHaveTextContent("2");
+ });
+
+ it("suffixes counts that reached the cap with '+'", () => {
+ renderCounts({ failed: 5, queued: 0, running: 1, success: 1000 }, {
stateCountLimit: 1000 });
+
expect(screen.getByTestId("run-state-count-success-my_dag")).toHaveTextContent("1000+");
+
expect(screen.getByTestId("run-state-count-failed-my_dag")).toHaveTextContent("5");
+
expect(screen.getByTestId("run-state-count-running-my_dag")).toHaveTextContent("1");
+ });
+});
diff --git
a/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.tsx
b/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.tsx
new file mode 100644
index 00000000000..4a238795ce1
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagRunStateCounts.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, Skeleton } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+
+import type { DagRunState } from "openapi/requests/types.gen";
+import { StateBadge } from "src/components/StateBadge";
+import { RouterLink, Tooltip } from "src/components/ui";
+import { SearchParamsKeys } from "src/constants/searchParams";
+
+const DISPLAYED_STATES: ReadonlyArray<DagRunState> = ["success", "failed",
"running", "queued"];
+
+type Props = {
+ readonly compact?: boolean;
+ readonly counts: Record<string, number> | undefined;
+ readonly dagId: string;
+ readonly isLoading: boolean;
+ readonly stateCountLimit: number | undefined;
+};
+
+export const DagRunStateCounts = ({ compact = false, counts, dagId, isLoading,
stateCountLimit }: Props) => {
+ const { t: translate } = useTranslation(["dags", "common"]);
+ const gap = compact ? 0.5 : 1;
+ const fontSize = compact ? "xs" : "sm";
+
+ if (isLoading || counts === undefined) {
+ // Render skeletons sized to a typical badge so the row doesn't reflow when
+ // counts arrive. Rendering zeros while loading would mislead the operator.
+ return (
+ <HStack
+ aria-label={translate("runStateCounts.loading")}
+ data-testid={`run-state-counts-loading-${dagId}`}
+ gap={gap}
+ >
+ {DISPLAYED_STATES.map((state) => (
+ <Skeleton borderRadius="full" height="22px" key={state} width="44px"
/>
+ ))}
+ </HStack>
+ );
+ }
+
+ return (
+ <HStack data-testid={`run-state-counts-${dagId}`} gap={gap}>
+ {DISPLAYED_STATES.map((state) => {
+ const count = counts[state] ?? 0;
+ // A count that reached the API cap is only a lower bound; suffix it
with "+".
+ const isCapped = stateCountLimit !== undefined && count >=
stateCountLimit;
+ const suffix = isCapped ? "+" : "";
+ const translatedState = translate(`common:states.${state}` as const);
+ const tooltipContent = translate("runStateCounts.tooltip", {
+ formattedCount: `${count}${suffix}`,
+ state: translatedState,
+ });
+
+ return (
+ <Tooltip content={tooltipContent} key={state}>
+ <RouterLink
+ aria-label={tooltipContent}
+ data-testid={`run-state-count-${state}-${dagId}`}
+ to={`/dags/${dagId}/runs?${SearchParamsKeys.STATE}=${state}`}
+ >
+ <StateBadge fontSize={fontSize} opacity={count === 0 ? 0.4 : 1}
state={state}>
+ {`${count}${suffix}`}
+ </StateBadge>
+ </RouterLink>
+ </Tooltip>
+ );
+ })}
+ </HStack>
+ );
+};
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 eaedd94406a..12e08fdb51b 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx
@@ -40,18 +40,27 @@ import { SearchParamsKeys, type SearchParamsKeysType } from
"src/constants/searc
import { useAdvancedSearch } from "src/hooks/useAdvancedSearch";
import { DagsLayout } from "src/layouts/DagsLayout";
import { useConfig } from "src/queries/useConfig";
+import { useDagRunStateCounts } from "src/queries/useDagRunStateCounts";
import { useDags } from "src/queries/useDags";
import { DagImportErrors } from "../Dashboard/Stats/DagImportErrors";
import { DagCard } from "./DagCard";
+import { DagRunStateCounts } from "./DagRunStateCounts";
import { DagTags } from "./DagTags";
import { DagsFilters } from "./DagsFilters";
import { Schedule } from "./Schedule";
import { SortSelect } from "./SortSelect";
import { useTagFilter } from "./useTagFilter";
+type RunStateCountsContext = {
+ readonly countsByDag: Record<string, Record<string, number> | undefined>;
+ readonly isLoading: boolean;
+ readonly stateCountLimit: number | undefined;
+};
+
const createColumns = (
translate: (key: string, options?: Record<string, unknown>) => string,
+ runStateContext: RunStateCountsContext,
): Array<ColumnDef<DAGWithLatestDagRunsResponse>> => [
{
accessorKey: "is_paused",
@@ -121,6 +130,20 @@ const createColumns = (
) : undefined,
header: () => translate("dagDetails.latestRun"),
},
+ {
+ accessorKey: "run_state_counts",
+ cell: ({ row: { original } }) => (
+ <DagRunStateCounts
+ compact
+ counts={runStateContext.countsByDag[original.dag_id]}
+ dagId={original.dag_id}
+ isLoading={runStateContext.isLoading}
+ stateCountLimit={runStateContext.stateCountLimit}
+ />
+ ),
+ enableSorting: false,
+ header: () => translate("dags:runStateCounts.label"),
+ },
{
accessorKey: "tags",
cell: ({
@@ -179,12 +202,19 @@ const {
PAUSED,
}: SearchParamsKeysType = SearchParamsKeys;
-const cardDef: CardDef<DAGWithLatestDagRunsResponse> = {
- card: ({ row }) => <DagCard dag={row} />,
+const createCardDef = (runStateContext: RunStateCountsContext):
CardDef<DAGWithLatestDagRunsResponse> => ({
+ card: ({ row }) => (
+ <DagCard
+ dag={row}
+ runStateCounts={runStateContext.countsByDag[row.dag_id]}
+ runStateCountsLoading={runStateContext.isLoading}
+ stateCountLimit={runStateContext.stateCountLimit}
+ />
+ ),
meta: {
- customSkeleton: <Skeleton height="120px" width="100%" />,
+ customSkeleton: <Skeleton height="140px" width="100%" />,
},
-};
+});
export const DagsList = () => {
const { t: translate } = useTranslation();
@@ -212,8 +242,6 @@ export const DagsList = () => {
const [sort] = sorting;
const orderBy = sort ? `${sort.desc ? "-" : ""}${sort.id}` :
"dag_display_name";
- const columns = createColumns(translate);
-
const handleSearchChange = (value: string) => {
setTableURLState({
pagination: { ...pagination, pageIndex: 0 },
@@ -268,6 +296,21 @@ export const DagsList = () => {
tagsMatchMode: selectedMatchMode,
});
+ const { data: runStateCountsData, isLoading: runStateCountsLoading } =
useDagRunStateCounts({
+ dagIds: data?.dags.map((dag) => dag.dag_id) ?? [],
+ dags: data?.dags,
+ });
+ const runStateContext: RunStateCountsContext = {
+ countsByDag: Object.fromEntries(
+ (runStateCountsData?.dags ?? []).map((entry) => [entry.dag_id,
entry.state_counts]),
+ ),
+ isLoading: runStateCountsLoading,
+ stateCountLimit: runStateCountsData?.state_count_limit,
+ };
+
+ const columns = createColumns(translate, runStateContext);
+ const cardDef = createCardDef(runStateContext);
+
const handleSortChange = ({ value }:
SelectValueChangeDetails<Array<string>>) => {
setTableURLState({
pagination,
@@ -297,9 +340,11 @@ export const DagsList = () => {
</Heading>
<DagImportErrors iconOnly />
</HStack>
- {display === "card" ? (
- <SortSelect handleSortChange={handleSortChange} orderBy={orderBy}
/>
- ) : undefined}
+ <HStack>
+ {display === "card" ? (
+ <SortSelect handleSortChange={handleSortChange}
orderBy={orderBy} />
+ ) : undefined}
+ </HStack>
</HStack>
</VStack>
<Box pb={8}>
diff --git a/airflow-core/src/airflow/ui/src/queries/useDagRunStateCounts.tsx
b/airflow-core/src/airflow/ui/src/queries/useDagRunStateCounts.tsx
new file mode 100644
index 00000000000..4e152fecfec
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/queries/useDagRunStateCounts.tsx
@@ -0,0 +1,45 @@
+/*!
+ * 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 { useDagServiceGetDagRunStateCountsUi } from "openapi/queries";
+import type { DAGWithLatestDagRunsResponse } from "openapi/requests/types.gen";
+import { isStatePending, useAutoRefresh } from "src/utils";
+
+export const useDagRunStateCounts = ({
+ dagIds,
+ dags,
+}: {
+ readonly dagIds: ReadonlyArray<string>;
+ // Refresh predicate is derived from useDags' data so the counts query
doesn't
+ // need to be loaded before it knows whether to poll — avoids a
chicken-and-egg.
+ readonly dags: ReadonlyArray<DAGWithLatestDagRunsResponse> | undefined;
+}) => {
+ const refetchInterval = useAutoRefresh({});
+ const hasPendingRun =
+ dags?.some((dag) => !dag.is_paused && dag.latest_dag_runs.some((run) =>
isStatePending(run.state))) ??
+ false;
+
+ // Stable key: sort the dag_ids so pagination/sort order changes don't churn
the cache.
+ const sortedDagIds = [...dagIds].sort();
+
+ return useDagServiceGetDagRunStateCountsUi({ dagIds: sortedDagIds },
undefined, {
+ enabled: sortedDagIds.length > 0,
+ placeholderData: (prev) => prev,
+ refetchInterval: hasPendingRun ? refetchInterval : false,
+ });
+};
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 bb94168bd57..50444e04914 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
@@ -26,6 +26,7 @@ from sqlalchemy.orm import Session
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
from airflow.api_fastapi.auth.managers.simple.simple_auth_manager import
SimpleAuthManager
+from airflow.configuration import conf
from airflow.models import DagRun
from airflow.models.dag import DagModel, DagTag
from airflow.models.dag_favorite import DagFavorite
@@ -418,3 +419,145 @@ class TestGetDagRuns(TestPublicDagEndpoint):
# Verify that DAG1 is not marked as favorite for the test user
dag1_data = next(dag for dag in body["dags"] if dag["dag_id"] ==
DAG1_ID)
assert dag1_data["is_favorite"] is False
+
+
+# Rows seeded by the parent ``TestPublicDagEndpoint.setup`` fixture. Tracked
here
+# so expected counts below stay derivable if the parent fixture is ever
changed.
+_PARENT_DAG1_FAILED = 1 # via ``dag_maker.create_dagrun(state=FAILED)``
+_PARENT_DAG3_FAILED = 1 # via ``_create_deactivated_paused_dag``
+_PARENT_DAG3_SUCCESS = 1 # via ``_create_deactivated_paused_dag``
+
+
+class TestGetDagRunStateCounts(TestPublicDagEndpoint):
+ """Tests for ``GET /ui/dags/run_state_counts``."""
+
+ @pytest.fixture(autouse=True)
+ def seed_dag_runs(self, setup, session) -> None:
+ # DAG1 gets runs in every state. DAG2 gets only SUCCESS/FAILED.
+ # DAG3 is left to the parent fixture.
+ base = utcnow() - pendulum.duration(days=10)
+ for dag_id, scheme in (
+ (
+ DAG1_ID,
+ [
+ DagRunState.SUCCESS,
+ DagRunState.SUCCESS,
+ DagRunState.FAILED,
+ DagRunState.FAILED,
+ DagRunState.RUNNING,
+ DagRunState.QUEUED,
+ ],
+ ),
+ (DAG2_ID, [DagRunState.SUCCESS, DagRunState.FAILED]),
+ ):
+ for idx, state in enumerate(scheme):
+ # Offset by idx seconds so the (dag_id, logical_date) and
+ # (dag_id, run_after) uniqueness constraints both hold.
+ run_after = base + pendulum.duration(seconds=idx)
+ session.add(
+ DagRun(
+ dag_id=dag_id,
+ run_id=f"{dag_id}_state_run_{idx}",
+ run_type=DagRunType.MANUAL,
+ start_date=run_after,
+ logical_date=run_after,
+ run_after=run_after,
+ state=state,
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ )
+ session.commit()
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_returns_zero_filled_counts_per_requested_dag(self, test_client):
+ response = test_client.get(
+ "/dags/run_state_counts",
+ params={"dag_ids": [DAG1_ID, DAG3_ID]},
+ )
+ assert response.status_code == 200
+ counts = {entry["dag_id"]: entry["state_counts"] for entry in
response.json()["dags"]}
+ # DAG1: 2 SUCCESS + 2 FAILED + 1 RUNNING + 1 QUEUED locally + parent
seeds.
+ assert counts[DAG1_ID] == {
+ "success": 2,
+ "failed": 2 + _PARENT_DAG1_FAILED,
+ "running": 1,
+ "queued": 1,
+ }
+ # DAG3: only parent seeds (this fixture skips DAG3).
+ assert counts[DAG3_ID] == {
+ "success": _PARENT_DAG3_SUCCESS,
+ "failed": _PARENT_DAG3_FAILED,
+ "running": 0,
+ "queued": 0,
+ }
+
+ def test_deduplicates_dag_ids(self, test_client):
+ response = test_client.get(
+ "/dags/run_state_counts",
+ params={"dag_ids": [DAG1_ID, DAG1_ID]},
+ )
+ assert response.status_code == 200
+ dag_ids = [entry["dag_id"] for entry in response.json()["dags"]]
+ assert dag_ids == [DAG1_ID]
+
+ def test_rejects_too_many_dag_ids(self, test_client):
+ # The page never sends more than maximum_page_limit dag_ids; a direct
call with a
+ # larger list is rejected so the per-Dag UNION ALL width stays bounded.
+ too_many = [f"dag_{idx}" for idx in range(conf.getint("api",
"maximum_page_limit") + 1)]
+ response = test_client.get("/dags/run_state_counts",
params={"dag_ids": too_many})
+ assert response.status_code == 422
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_permission_filter_hides_disallowed_dags(self, test_client,
session):
+ # The test user is granted read on DAG1, DAG2, DAG4, DAG5 (DAG3 is
paused/stale in
+ # the parent fixture). Asking for a dag that exists but the caller
cannot read
+ # should silently drop it from the result.
+ with mock.patch.object(
+ SimpleAuthManager,
+ "get_authorized_dag_ids",
+ return_value={DAG1_ID},
+ ):
+ response = test_client.get(
+ "/dags/run_state_counts",
+ params={"dag_ids": [DAG1_ID, DAG2_ID]},
+ )
+ assert response.status_code == 200
+ dag_ids = [entry["dag_id"] for entry in response.json()["dags"]]
+ assert dag_ids == [DAG1_ID]
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ @mock.patch("airflow.api_fastapi.core_api.routes.ui.dags.STATE_COUNT_CAP",
3)
+ def test_caps_counts_at_state_count_cap(self, test_client, session):
+ # Push SUCCESS over the (patched) cap while FAILED stays under it:
states cap independently.
+ base = utcnow() - pendulum.duration(days=5)
+ for idx in range(5):
+ run_after = base + pendulum.duration(seconds=idx)
+ session.add(
+ DagRun(
+ dag_id=DAG2_ID,
+ run_id=f"{DAG2_ID}_cap_run_{idx}",
+ run_type=DagRunType.MANUAL,
+ start_date=run_after,
+ logical_date=run_after,
+ run_after=run_after,
+ state=DagRunState.SUCCESS,
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ )
+ session.commit()
+
+ response = test_client.get("/dags/run_state_counts",
params={"dag_ids": [DAG2_ID]})
+ assert response.status_code == 200
+ body = response.json()
+ assert body["state_count_limit"] == 3
+ counts = {entry["dag_id"]: entry["state_counts"] for entry in
body["dags"]}
+ assert counts[DAG2_ID]["success"] == 3
+ assert counts[DAG2_ID]["failed"] == 1
+
+ def test_should_response_401(self, unauthenticated_test_client):
+ response = unauthenticated_test_client.get("/dags/run_state_counts",
params={"dag_ids": [DAG1_ID]})
+ assert response.status_code == 401
+
+ def test_should_response_403(self, unauthorized_test_client):
+ response = unauthorized_test_client.get("/dags/run_state_counts",
params={"dag_ids": [DAG1_ID]})
+ assert response.status_code == 403
