Add dry run for backfill

airflow/api_fastapi/core_api/datamodels/backfills.py

@@ -33,6 +33,7 @@ class BackfillPostBody(BaseModel):
     dag_run_conf: dict = {}
     reprocess_behavior: ReprocessBehavior = ReprocessBehavior.NONE
     max_active_runs: int = 10
+    dry_run: bool = False
 
 
 class BackfillResponse(BaseModel):
@@ -56,3 +57,15 @@ class BackfillCollectionResponse(BaseModel):
 
     backfills: list[BackfillResponse]
     total_entries: int
+
+
+class BackfillRunInfo(BaseModel):
+    """Data model for run information during a backfill operation."""
+
+    logical_date: datetime
+
+
+class BackfillDryRunResponse(BaseModel):
+    """Serializer for responses in dry-run mode for backfill operations."""
+
+    run_info_list: list[BackfillRunInfo]

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -1365,6 +1365,55 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /public/backfills/dry_run:
+    post:
+      tags:
+      - Backfill
+      summary: Create Backfill Dry Run
+      operationId: create_backfill_dry_run
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/BackfillPostBody'
+        required: true
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/BackfillDryRunResponse'
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '403':
+          description: Forbidden
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '404':
+          description: Not Found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '409':
+          description: Conflict
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /public/connections/{connection_id}:
     delete:
       tags:
@@ -6216,6 +6265,18 @@ components:
       - total_entries
       title: BackfillCollectionResponse
       description: Backfill Collection serializer for responses.
+    BackfillDryRunResponse:
+      properties:
+        run_info_list:
+          items:
+            $ref: '#/components/schemas/BackfillRunInfo'
+          type: array
+          title: Run Info List
+      type: object
+      required:
+      - run_info_list
+      title: BackfillDryRunResponse
+      description: Serializer for responses in dry-run mode for backfill operations.
     BackfillPostBody:
       properties:
         dag_id:
@@ -6244,6 +6305,10 @@ components:
           type: integer
           title: Max Active Runs
           default: 10
+        dry_run:
+          type: boolean
+          title: Dry Run
+          default: false
       type: object
       required:
       - dag_id
@@ -6307,6 +6372,17 @@ components:
       - updated_at
       title: BackfillResponse
       description: Base serializer for Backfill.
+    BackfillRunInfo:
+      properties:
+        logical_date:
+          type: string
+          format: date-time
+          title: Logical Date
+      type: object
+      required:
+      - logical_date
+      title: BackfillRunInfo
+      description: Data model for run information during a backfill operation.
     BaseInfoResponse:
       properties:
         status:

airflow/api_fastapi/core_api/routes/public/backfills.py

@@ -19,7 +19,7 @@
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, status
-from sqlalchemy import select, update
+from sqlalchemy import desc, select, update
 
 from airflow.api_fastapi.common.db.common import (
     AsyncSessionDep,
@@ -30,8 +30,10 @@
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.datamodels.backfills import (
     BackfillCollectionResponse,
+    BackfillDryRunResponse,
     BackfillPostBody,
     BackfillResponse,
+    BackfillRunInfo,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
@@ -41,9 +43,14 @@
     AlreadyRunningBackfill,
     Backfill,
     BackfillDagRun,
+    BackfillDagRunExceptionReason,
+    ReprocessBehavior,
     _create_backfill,
+    _get_info_list,
 )
+from airflow.models.serialized_dag import SerializedDagModel
 from airflow.utils import timezone
+from airflow.utils.sqlalchemy import nulls_first
 from airflow.utils.state import DagRunState
 
 backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
@@ -206,3 +213,55 @@ def create_backfill(
             status_code=status.HTTP_409_CONFLICT,
             detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
         )
+
+
+@backfills_router.post(
+    path="/dry_run",
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_404_NOT_FOUND,
+            status.HTTP_409_CONFLICT,
+        ]
+    ),
+)
+def create_backfill_dry_run(
+    backfill_request: BackfillPostBody,
+    session: SessionDep,
+) -> BackfillDryRunResponse:
+    from_date = timezone.coerce_datetime(backfill_request.from_date)
+    to_date = timezone.coerce_datetime(backfill_request.to_date)
+    serdag = session.scalar(SerializedDagModel.latest_item_select_object(backfill_request.dag_id))
+    if not serdag:
+        raise HTTPException(status_code=404, detail=f"Could not find dag {backfill_request.dag_id}")
+
+    info_list = _get_info_list(
+        dag=serdag.dag,
+        from_date=from_date,
+        to_date=to_date,
+        reverse=backfill_request.run_backwards,
+    )
+    backfill_response_item = []
+    for info in info_list:
+        dr = session.scalar(
+            select(DagRun)
+            .where(DagRun.logical_date == info.logical_date)
+            .order_by(nulls_first(desc(DagRun.start_date), session))
+            .limit(1)
+        )
+
+        if dr:
+            non_create_reason = None
+            if dr.state not in (DagRunState.SUCCESS, DagRunState.FAILED):
+                non_create_reason = BackfillDagRunExceptionReason.IN_FLIGHT
+            elif backfill_request.reprocess_behavior is ReprocessBehavior.NONE:
+                non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
+            elif backfill_request.reprocess_behavior is ReprocessBehavior.FAILED:
+                if dr.state != DagRunState.FAILED:
+                    non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
+            if not non_create_reason:
+                backfill_response_item.append(BackfillRunInfo(logical_date=info.logical_date))
+
+        else:
+            backfill_response_item.append(BackfillRunInfo(logical_date=info.logical_date))
+
+    return BackfillDryRunResponse(run_info_list=backfill_response_item)

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

@@ -1686,6 +1686,9 @@ export type AssetServiceCreateAssetEventMutationResult = Awaited<
 export type BackfillServiceCreateBackfillMutationResult = Awaited<
   ReturnType<typeof BackfillService.createBackfill>
 >;
+export type BackfillServiceCreateBackfillDryRunMutationResult = Awaited<
+  ReturnType<typeof BackfillService.createBackfillDryRun>
+>;
 export type ConnectionServicePostConnectionMutationResult = Awaited<
   ReturnType<typeof ConnectionService.postConnection>
 >;

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

@@ -2813,6 +2813,44 @@ export const useBackfillServiceCreateBackfill = <
       }) as unknown as Promise<TData>,
     ...options,
   });
+/**
+ * Create Backfill Dry Run
+ * @param data The data for the request.
+ * @param data.requestBody
+ * @returns BackfillDryRunResponse Successful Response
+ * @throws ApiError
+ */
+export const useBackfillServiceCreateBackfillDryRun = <
+  TData = Common.BackfillServiceCreateBackfillDryRunMutationResult,
+  TError = unknown,
+  TContext = unknown,
+>(
+  options?: Omit<
+    UseMutationOptions<
+      TData,
+      TError,
+      {
+        requestBody: BackfillPostBody;
+      },
+      TContext
+    >,
+    "mutationFn"
+  >,
+) =>
+  useMutation<
+    TData,
+    TError,
+    {
+      requestBody: BackfillPostBody;
+    },
+    TContext
+  >({
+    mutationFn: ({ requestBody }) =>
+      BackfillService.createBackfillDryRun({
+        requestBody,
+      }) as unknown as Promise<TData>,
+    ...options,
+  });
 /**
  * Post Connection
  * Create connection entry.

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

@@ -350,6 +350,23 @@ export const $BackfillCollectionResponse = {
   description: "Backfill Collection serializer for responses.",
 } as const;
 
+export const $BackfillDryRunResponse = {
+  properties: {
+    run_info_list: {
+      items: {
+        $ref: "#/components/schemas/BackfillRunInfo",
+      },
+      type: "array",
+      title: "Run Info List",
+    },
+  },
+  type: "object",
+  required: ["run_info_list"],
+  title: "BackfillDryRunResponse",
+  description:
+    "Serializer for responses in dry-run mode for backfill operations.",
+} as const;
+
 export const $BackfillPostBody = {
   properties: {
     dag_id: {
@@ -385,6 +402,11 @@ export const $BackfillPostBody = {
       title: "Max Active Runs",
       default: 10,
     },
+    dry_run: {
+      type: "boolean",
+      title: "Dry Run",
+      default: false,
+    },
   },
   type: "object",
   required: ["dag_id", "from_date", "to_date"],
@@ -468,6 +490,20 @@ export const $BackfillResponse = {
   description: "Base serializer for Backfill.",
 } as const;
 
+export const $BackfillRunInfo = {
+  properties: {
+    logical_date: {
+      type: "string",
+      format: "date-time",
+      title: "Logical Date",
+    },
+  },
+  type: "object",
+  required: ["logical_date"],
+  title: "BackfillRunInfo",
+  description: "Data model for run information during a backfill operation.",
+} as const;
+
 export const $BaseInfoResponse = {
   properties: {
     status: {

airflow/ui/openapi-gen/requests/services.gen.ts

@@ -54,6 +54,8 @@ import type {
   UnpauseBackfillResponse,
   CancelBackfillData,
   CancelBackfillResponse,
+  CreateBackfillDryRunData,
+  CreateBackfillDryRunResponse,
   DeleteConnectionData,
   DeleteConnectionResponse,
   GetConnectionData,
@@ -960,6 +962,31 @@ export class BackfillService {
       },
     });
   }
+
+  /**
+   * Create Backfill Dry Run
+   * @param data The data for the request.
+   * @param data.requestBody
+   * @returns BackfillDryRunResponse Successful Response
+   * @throws ApiError
+   */
+  public static createBackfillDryRun(
+    data: CreateBackfillDryRunData,
+  ): CancelablePromise<CreateBackfillDryRunResponse> {
+    return __request(OpenAPI, {
+      method: "POST",
+      url: "/public/backfills/dry_run",
+      body: data.requestBody,
+      mediaType: "application/json",
+      errors: {
+        401: "Unauthorized",
+        403: "Forbidden",
+        404: "Not Found",
+        409: "Conflict",
+        422: "Validation Error",
+      },
+    });
+  }
 }
 
 export class ConnectionService {