Show task state counts for the latest Dag run on the Dags list page

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

@@ -48,6 +48,25 @@ class DAGRunStateCountsResponse(BaseModel):
     state_counts: dict[DagRunState, int]
 
 
+class DAGLatestRunTaskInstanceStateCountsResponse(BaseModel):
+    """
+    Task-instance state counts for a Dag's latest run.
+
+    ``state_counts`` only carries states present in the run; task instances without a
+    state yet are keyed as ``no_status``.
+    """
+
+    dag_id: str
+    run_id: str
+    state_counts: dict[str, int]
+
+
+class DAGsLatestRunTaskInstanceStateCountsCollectionResponse(BaseModel):
+    """Collection of per-Dag latest-run task-instance state counts for the Dag list page."""
+
+    dags: list[DAGLatestRunTaskInstanceStateCountsResponse]
+
+
 class DAGsRunStateCountsCollectionResponse(BaseModel):
     """Collection of per-Dag DagRun-state counts for the Dag list page."""
 

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -640,6 +640,44 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /ui/dags/latest_run_task_instance_state_counts:
+    get:
+      tags:
+      - DAG
+      summary: Get Latest Run Task Instance State Counts
+      description: 'Return task-instance state counts for each Dag''s latest run,
+        for the Dag list page.
+
+
+        Dags without any run are omitted from the response.'
+      operationId: get_latest_run_task_instance_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/DAGsLatestRunTaskInstanceStateCountsCollectionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /ui/dependencies:
     get:
       tags:
@@ -2366,6 +2404,32 @@ components:
         that
 
         the API server/Web UI can use this data to render connection form UI.'
+    DAGLatestRunTaskInstanceStateCountsResponse:
+      properties:
+        dag_id:
+          type: string
+          title: Dag Id
+        run_id:
+          type: string
+          title: Run Id
+        state_counts:
+          additionalProperties:
+            type: integer
+          type: object
+          title: State Counts
+      type: object
+      required:
+      - dag_id
+      - run_id
+      - state_counts
+      title: DAGLatestRunTaskInstanceStateCountsResponse
+      description: 'Task-instance state counts for a Dag''s latest run.
+
+
+        ``state_counts`` only carries states present in the run; task instances without
+        a
+
+        state yet are keyed as ``no_status``.'
     DAGRunLightResponse:
       properties:
         id:
@@ -2677,6 +2741,19 @@ components:
       - file_token
       title: DAGWithLatestDagRunsResponse
       description: DAG with latest dag runs response serializer.
+    DAGsLatestRunTaskInstanceStateCountsCollectionResponse:
+      properties:
+        dags:
+          items:
+            $ref: '#/components/schemas/DAGLatestRunTaskInstanceStateCountsResponse'
+          type: array
+          title: Dags
+      type: object
+      required:
+      - dags
+      title: DAGsLatestRunTaskInstanceStateCountsCollectionResponse
+      description: Collection of per-Dag latest-run task-instance state counts for
+        the Dag list page.
     DAGsRunStateCountsCollectionResponse:
       properties:
         dags:

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

@@ -58,7 +58,9 @@
 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 (
+    DAGLatestRunTaskInstanceStateCountsResponse,
     DAGRunStateCountsResponse,
+    DAGsLatestRunTaskInstanceStateCountsCollectionResponse,
     DAGsRunStateCountsCollectionResponse,
     DAGWithLatestDagRunsCollectionResponse,
     DAGWithLatestDagRunsResponse,
@@ -343,3 +345,75 @@ def get_dag_run_state_counts(
         ],
         state_count_limit=STATE_COUNT_CAP,
     )
+
+
+@dags_router.get(
+    "/latest_run_task_instance_state_counts",
+    dependencies=[
+        Depends(requires_access_dag(method="GET")),
+        Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK_INSTANCE)),
+    ],
+    operation_id="get_latest_run_task_instance_state_counts_ui",
+)
+def get_latest_run_task_instance_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"))],
+) -> DAGsLatestRunTaskInstanceStateCountsCollectionResponse:
+    """
+    Return task-instance state counts for each Dag's latest run, for the Dag list page.
+
+    Dags without any run are omitted from the response.
+    """
+    permitted_dag_ids = readable_dags_filter.value or set()
+    requested_dag_ids = sorted(set(dag_ids) & permitted_dag_ids)
+
+    dags: list[DAGLatestRunTaskInstanceStateCountsResponse] = []
+    if not requested_dag_ids:
+        return DAGsLatestRunTaskInstanceStateCountsCollectionResponse(dags=dags)
+
+    latest_run_branches = [
+        select(DagRun.dag_id, DagRun.run_id)
+        .where(DagRun.dag_id == dag_id)
+        .order_by(DagRun.run_after.desc())
+        .limit(1)
+        .subquery()
+        for dag_id in requested_dag_ids
+    ]
+    latest_runs_union = union_all(*(select(branch) for branch in latest_run_branches)).subquery()
+    latest_run_id_by_dag: dict[str, str] = {
+        row.dag_id: row.run_id for row in session.execute(select(latest_runs_union))
+    }
+
+    if latest_run_id_by_dag:
+        # Each branch filters on (dag_id, run_id) equality, which the ti_dag_run index
+        # covers. A run's task instances are bounded by the Dag's task structure, so the
+        # per-state counts are exact (no cap needed here, unlike the cross-run counts in
+        # get_dag_run_state_counts).
+        ti_branches = [
+            select(literal(dag_id).label("dag_id"), TaskInstance.state.label("state"))
+            .where(TaskInstance.dag_id == dag_id, TaskInstance.run_id == run_id)
+            .subquery()
+            for dag_id, run_id in latest_run_id_by_dag.items()
+        ]
+        tis_union = union_all(*(select(branch) for branch in ti_branches)).subquery()
+        counts_by_dag: dict[str, dict[str, int]] = {dag_id: {} for dag_id in latest_run_id_by_dag}
+        for row in session.execute(
+            select(tis_union.c.dag_id, tis_union.c.state, func.count().label("cnt")).group_by(
+                tis_union.c.dag_id, tis_union.c.state
+            )
+        ):
+            state_key = row.state if row.state is not None else "no_status"
+            counts_by_dag[row.dag_id][state_key] = row.cnt
+
+        dags = [
+            DAGLatestRunTaskInstanceStateCountsResponse(
+                dag_id=dag_id,
+                run_id=latest_run_id_by_dag[dag_id],
+                state_counts=counts_by_dag[dag_id],
+            )
+            for dag_id in requested_dag_ids
+            if dag_id in latest_run_id_by_dag
+        ]
+
+    return DAGsLatestRunTaskInstanceStateCountsCollectionResponse(dags=dags)

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -364,6 +364,12 @@ export const useDagServiceGetDagRunStateCountsUiKey = "DagServiceGetDagRunStateC
 export const UseDagServiceGetDagRunStateCountsUiKeyFn = ({ dagIds }: {
   dagIds: string[];
 }, queryKey?: Array<unknown>) => [useDagServiceGetDagRunStateCountsUiKey, ...(queryKey ?? [{ dagIds }])];
+export type DagServiceGetLatestRunTaskInstanceStateCountsUiDefaultResponse = Awaited<ReturnType<typeof DagService.getLatestRunTaskInstanceStateCountsUi>>;
+export type DagServiceGetLatestRunTaskInstanceStateCountsUiQueryResult<TData = DagServiceGetLatestRunTaskInstanceStateCountsUiDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
+export const useDagServiceGetLatestRunTaskInstanceStateCountsUiKey = "DagServiceGetLatestRunTaskInstanceStateCountsUi";
+export const UseDagServiceGetLatestRunTaskInstanceStateCountsUiKeyFn = ({ dagIds }: {
+  dagIds: string[];
+}, queryKey?: Array<unknown>) => [useDagServiceGetLatestRunTaskInstanceStateCountsUiKey, ...(queryKey ?? [{ dagIds }])];
 export type EventLogServiceGetEventLogDefaultResponse = Awaited<ReturnType<typeof EventLogService.getEventLog>>;
 export type EventLogServiceGetEventLogQueryResult<TData = EventLogServiceGetEventLogDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useEventLogServiceGetEventLogKey = "EventLogServiceGetEventLog";

airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts

@@ -724,6 +724,19 @@ export const ensureUseDagServiceGetDagRunStateCountsUiData = (queryClient: Query
   dagIds: string[];
 }) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) });
 /**
+* Get Latest Run Task Instance State Counts
+* Return task-instance state counts for each Dag's latest run, for the Dag list page.
+*
+* Dags without any run are omitted from the response.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsLatestRunTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseDagServiceGetLatestRunTaskInstanceStateCountsUiData = (queryClient: QueryClient, { dagIds }: {
+  dagIds: string[];
+}) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetLatestRunTaskInstanceStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getLatestRunTaskInstanceStateCountsUi({ dagIds }) });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts

@@ -724,6 +724,19 @@ export const prefetchUseDagServiceGetDagRunStateCountsUi = (queryClient: QueryCl
   dagIds: string[];
 }) => queryClient.prefetchQuery({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) });
 /**
+* Get Latest Run Task Instance State Counts
+* Return task-instance state counts for each Dag's latest run, for the Dag list page.
+*
+* Dags without any run are omitted from the response.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsLatestRunTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseDagServiceGetLatestRunTaskInstanceStateCountsUi = (queryClient: QueryClient, { dagIds }: {
+  dagIds: string[];
+}) => queryClient.prefetchQuery({ queryKey: Common.UseDagServiceGetLatestRunTaskInstanceStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getLatestRunTaskInstanceStateCountsUi({ dagIds }) });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts

@@ -724,6 +724,19 @@ export const useDagServiceGetDagRunStateCountsUi = <TData = Common.DagServiceGet
   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 Latest Run Task Instance State Counts
+* Return task-instance state counts for each Dag's latest run, for the Dag list page.
+*
+* Dags without any run are omitted from the response.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsLatestRunTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetLatestRunTaskInstanceStateCountsUi = <TData = Common.DagServiceGetLatestRunTaskInstanceStateCountsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagIds }: {
+  dagIds: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseDagServiceGetLatestRunTaskInstanceStateCountsUiKeyFn({ dagIds }, queryKey), queryFn: () => DagService.getLatestRunTaskInstanceStateCountsUi({ dagIds }) as TData, ...options });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts

@@ -724,6 +724,19 @@ export const useDagServiceGetDagRunStateCountsUiSuspense = <TData = Common.DagSe
   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 Latest Run Task Instance State Counts
+* Return task-instance state counts for each Dag's latest run, for the Dag list page.
+*
+* Dags without any run are omitted from the response.
+* @param data The data for the request.
+* @param data.dagIds
+* @returns DAGsLatestRunTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetLatestRunTaskInstanceStateCountsUiSuspense = <TData = Common.DagServiceGetLatestRunTaskInstanceStateCountsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagIds }: {
+  dagIds: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseDagServiceGetLatestRunTaskInstanceStateCountsUiKeyFn({ dagIds }, queryKey), queryFn: () => DagService.getLatestRunTaskInstanceStateCountsUi({ dagIds }) as TData, ...options });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

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

@@ -8865,6 +8865,33 @@ It is used to transfer providers information loaded by providers_manager such th
 the API server/Web UI can use this data to render connection form UI.`
 } as const;
 
+export const $DAGLatestRunTaskInstanceStateCountsResponse = {
+    properties: {
+        dag_id: {
+            type: 'string',
+            title: 'Dag Id'
+        },
+        run_id: {
+            type: 'string',
+            title: 'Run Id'
+        },
+        state_counts: {
+            additionalProperties: {
+                type: 'integer'
+            },
+            type: 'object',
+            title: 'State Counts'
+        }
+    },
+    type: 'object',
+    required: ['dag_id', 'run_id', 'state_counts'],
+    title: 'DAGLatestRunTaskInstanceStateCountsResponse',
+    description: `Task-instance state counts for a Dag's latest run.
+
+\`\`state_counts\`\` only carries states present in the run; task instances without a
+state yet are keyed as \`\`no_status\`\`.`
+} as const;
+
 export const $DAGRunLightResponse = {
     properties: {
         id: {
@@ -9315,6 +9342,22 @@ export const $DAGWithLatestDagRunsResponse = {
     description: 'DAG with latest dag runs response serializer.'
 } as const;
 
+export const $DAGsLatestRunTaskInstanceStateCountsCollectionResponse = {
+    properties: {
+        dags: {
+            items: {
+                '$ref': '#/components/schemas/DAGLatestRunTaskInstanceStateCountsResponse'
+            },
+            type: 'array',
+            title: 'Dags'
+        }
+    },
+    type: 'object',
+    required: ['dags'],
+    title: 'DAGsLatestRunTaskInstanceStateCountsCollectionResponse',
+    description: 'Collection of per-Dag latest-run task-instance state counts for the Dag list page.'
+} as const;
+
 export const $DAGsRunStateCountsCollectionResponse = {
     properties: {
         dags: {