feat: Add dagrun status progress to backfills

Author: shivaam

airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py

@@ -54,6 +54,8 @@ class BackfillResponse(BaseModel):
     completed_at: datetime | None
     updated_at: datetime
     dag_display_name: str = Field(validation_alias=AliasPath("dag_model", "dag_display_name"))
+    num_runs: int = 0
+    dag_run_state_counts: dict[str, int] = {}
 
 
 class BackfillCollectionResponse(BaseModel):

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

@@ -1483,6 +1483,16 @@ components:
         dag_display_name:
           type: string
           title: Dag Display Name
+        num_runs:
+          type: integer
+          title: Num Runs
+          default: 0
+        dag_run_state_counts:
+          additionalProperties:
+            type: integer
+          type: object
+          title: Dag Run State Counts
+          default: {}
       type: object
       required:
       - id

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -9296,6 +9296,16 @@ components:
         dag_display_name:
           type: string
           title: Dag Display Name
+        num_runs:
+          type: integer
+          title: Num Runs
+          default: 0
+        dag_run_state_counts:
+          additionalProperties:
+            type: integer
+          type: object
+          title: Dag Run State Counts
+          default: {}
       type: object
       required:
       - id

airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py

@@ -16,13 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+from collections import defaultdict
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import NonNegativeInt
-from sqlalchemy import select, update
-from sqlalchemy.orm import joinedload
+from sqlalchemy import func, select, update
+from sqlalchemy.orm import Session, joinedload
 
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.common.db.common import (
@@ -61,6 +62,40 @@
 backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
 
 
+def _enrich_backfill_responses(
+    backfills: list[BackfillResponse],
+    *,
+    session: Session,
+) -> list[BackfillResponse]:
+    """Populate num_runs and dag_run_state_counts on each backfill response."""
+    ids = [b.id for b in backfills]
+    if not ids:
+        return backfills
+    # Single query: get state counts per backfill, derive num_runs by summing counts.
+    rows = session.execute(
+        select(
+            BackfillDagRun.backfill_id,
+            DagRun.state,
+            func.count().label("count"),
+        )
+        .join(DagRun, BackfillDagRun.dag_run_id == DagRun.id)
+        .where(
+            BackfillDagRun.backfill_id.in_(ids),
+            DagRun.backfill_id == BackfillDagRun.backfill_id,
+        )
+        .group_by(BackfillDagRun.backfill_id, DagRun.state)
+    ).all()
+    counts: dict[int, dict[str, int]] = defaultdict(dict)
+    num_runs: dict[int, int] = defaultdict(int)
+    for backfill_id, state, count in rows:
+        counts[backfill_id][state] = count
+        num_runs[backfill_id] += count
+    for backfill in backfills:
+        backfill.num_runs = num_runs.get(backfill.id, 0)
+        backfill.dag_run_state_counts = counts.get(backfill.id, {})
+    return backfills
+
+
 @backfills_router.get(
     path="",
     dependencies=[
@@ -84,8 +119,10 @@ def list_backfills(
         limit=limit,
         session=session,
     )
+    backfills = [BackfillResponse.model_validate(b) for b in session.scalars(select_stmt)]
+    _enrich_backfill_responses(backfills, session=session)
     return BackfillCollectionResponse(
-        backfills=session.scalars(select_stmt),
+        backfills=backfills,
         total_entries=total_entries,
     )
 
@@ -104,9 +141,11 @@ def get_backfill(
     backfill = session.scalars(
         select(Backfill).where(Backfill.id == backfill_id).options(joinedload(Backfill.dag_model))
     ).one_or_none()
-    if backfill:
-        return backfill
-    raise HTTPException(status.HTTP_404_NOT_FOUND, "Backfill not found")
+    if not backfill:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Backfill not found")
+    response = BackfillResponse.model_validate(backfill)
+    _enrich_backfill_responses([response], session=session)
+    return response
 
 
 @backfills_router.put(
@@ -133,7 +172,9 @@ def pause_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> Backfill
     if b.is_paused is False:
         b.is_paused = True
     session.commit()
-    return b
+    response = BackfillResponse.model_validate(b)
+    _enrich_backfill_responses([response], session=session)
+    return response
 
 
 @backfills_router.put(
@@ -159,7 +200,9 @@ def unpause_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> Backfi
         raise HTTPException(status.HTTP_409_CONFLICT, "Backfill is already completed.")
     if b.is_paused:
         b.is_paused = False
-    return b
+    response = BackfillResponse.model_validate(b)
+    _enrich_backfill_responses([response], session=session)
+    return response
 
 
 @backfills_router.put(
@@ -210,7 +253,9 @@ def cancel_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> Backfil
     # this is in separate transaction just to avoid potential conflicts
     session.refresh(b)
     b.completed_at = timezone.utcnow()
-    return b
+    response = BackfillResponse.model_validate(b)
+    _enrich_backfill_responses([response], session=session)
+    return response
 
 
 @backfills_router.post(

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

@@ -36,6 +36,7 @@
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
 )
+from airflow.api_fastapi.core_api.routes.public.backfills import _enrich_backfill_responses
 from airflow.api_fastapi.core_api.security import ReadableBackfillsFilterDep, requires_access_backfill
 from airflow.models.backfill import Backfill
 
@@ -73,9 +74,12 @@ def list_backfills_ui(
         session=session,
     )
     backfills = [
-        BackfillResponse(**row._mapping) if not isinstance(row, Backfill) else row
+        BackfillResponse(**row._mapping)
+        if not isinstance(row, Backfill)
+        else BackfillResponse.model_validate(row)
         for row in session.scalars(select_stmt)
     ]
+    _enrich_backfill_responses(backfills, session=session)
     return BackfillCollectionResponse(
         backfills=backfills,
         total_entries=total_entries,

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

@@ -541,6 +541,19 @@ export const $BackfillResponse = {
         dag_display_name: {
             type: 'string',
             title: 'Dag Display Name'
+        },
+        num_runs: {
+            type: 'integer',
+            title: 'Num Runs',
+            default: 0
+        },
+        dag_run_state_counts: {
+            additionalProperties: {
+                type: 'integer'
+            },
+            type: 'object',
+            title: 'Dag Run State Counts',
+            default: {}
         }
     },
     type: 'object',

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

@@ -147,6 +147,10 @@ export type BackfillResponse = {
     completed_at: string | null;
     updated_at: string;
     dag_display_name: string;
+    num_runs?: number;
+    dag_run_state_counts?: {
+        [key: string]: (number);
+    };
 };
 
 /**

airflow-core/src/airflow/ui/public/i18n/locales/en/common.json

@@ -28,6 +28,7 @@
   },
   "collapseAllExtra": "Collapse all extra JSON",
   "collapseDetailsPanel": "Collapse Details Panel",
+  "completed": "Completed",
   "createdAssetEvent_one": "Created Asset Event",
   "createdAssetEvent_other": "Created Asset Events",
   "dag_one": "Dag",
@@ -232,6 +233,7 @@
     "from": "From",
     "maxActiveRuns": "Max Active Runs",
     "noTagsFound": "No tags found",
+    "progress": "Progress",
     "tagMode": {
       "all": "All",
       "any": "Any"

airflow-core/src/airflow/ui/src/components/BackfillProgressBar.tsx

@@ -0,0 +1,63 @@
+/*!
+ * 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 { Box, HStack, Text } from "@chakra-ui/react";
+
+type Props = {
+  readonly stateCounts: Record<string, number>;
+  readonly total: number;
+  readonly trackColor?: string;
+};
+
+export const BackfillProgressBar = ({ stateCounts, total, trackColor = "bg.emphasized" }: Props) => {
+  const successCount = stateCounts.success ?? 0;
+  const failedCount = stateCounts.failed ?? 0;
+  const successPct = (successCount / total) * 100;
+  const failedPct = (failedCount / total) * 100;
+  const remainingPct = 100 - successPct - failedPct;
+
+  return (
+    <HStack gap="2" minWidth="60px">
+      <HStack flex="1" gap={0}>
+        {successPct > 0 ? (
+          <Box bg="success.solid" borderLeftRadius={5} height="5px" width={`${successPct}%`} />
+        ) : undefined}
+        {failedPct > 0 ? (
+          <Box
+            bg="failed.solid"
+            borderLeftRadius={successCount === 0 ? 5 : 0}
+            height="5px"
+            width={`${failedPct}%`}
+          />
+        ) : undefined}
+        {remainingPct > 0 ? (
+          <Box
+            bg={trackColor}
+            borderLeftRadius={successCount === 0 && failedCount === 0 ? 5 : 0}
+            borderRightRadius={5}
+            height="5px"
+            width={`${remainingPct}%`}
+          />
+        ) : undefined}
+      </HStack>
+      <Text fontSize="sm" whiteSpace="nowrap">
+        {successCount + failedCount}/{total}
+      </Text>
+    </HStack>
+  );
+};

airflow-core/src/airflow/ui/src/components/Banner/BackfillBanner.tsx

@@ -32,8 +32,8 @@ import {
 import type { BackfillResponse } from "openapi/requests/types.gen";
 import { useAutoRefresh } from "src/utils";
 
+import { BackfillProgressBar } from "../BackfillProgressBar";
 import Time from "../Time";
-import { ProgressBar } from "../ui";
 
 type Props = {
   readonly dagId: string;
@@ -113,7 +113,13 @@ const BackfillBanner = ({ dagId }: Props) => {
         </Text>
 
         <Spacer flex="max-content" />
-        <ProgressBar size="xs" visibility="visible" />
+        {(backfill.num_runs ?? 0) > 0 ? (
+          <BackfillProgressBar
+            stateCounts={backfill.dag_run_state_counts ?? {}}
+            total={backfill.num_runs ?? 0}
+            trackColor="whiteAlpha.400"
+          />
+        ) : undefined}
         <Button
           aria-label={backfill.is_paused ? translate("banner.unpause") : translate("banner.pause")}
           loading={isPausePending || isUnPausePending}