Add dry run for backfill

airflow/api_fastapi/core_api/datamodels/backfills.py

@@ -56,3 +56,16 @@ class BackfillCollectionResponse(BaseModel):
 
     backfills: list[BackfillResponse]
     total_entries: int
+
+
+class DryRunBackfillResponse(BaseModel):
+    """Data model for run information during a backfill operation."""
+
+    logical_date: datetime
+
+
+class DryRunBackfillCollectionResponse(BaseModel):
+    """Serializer for responses in dry-run mode for backfill operations."""
+
+    backfills: list[DryRunBackfillResponse]
+    total_entries: int

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

@@ -1481,6 +1481,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/DryRunBackfillCollectionResponse'
+        '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:
@@ -7912,6 +7961,33 @@ components:
         This is the set of allowable values for the ``warning_type`` field
 
         in the DagWarning model.'
+    DryRunBackfillCollectionResponse:
+      properties:
+        backfills:
+          items:
+            $ref: '#/components/schemas/DryRunBackfillResponse'
+          type: array
+          title: Backfills
+        total_entries:
+          type: integer
+          title: Total Entries
+      type: object
+      required:
+      - backfills
+      - total_entries
+      title: DryRunBackfillCollectionResponse
+      description: Serializer for responses in dry-run mode for backfill operations.
+    DryRunBackfillResponse:
+      properties:
+        logical_date:
+          type: string
+          format: date-time
+          title: Logical Date
+      type: object
+      required:
+      - logical_date
+      title: DryRunBackfillResponse
+      description: Data model for run information during a backfill operation.
     EdgeResponse:
       properties:
         is_setup_teardown:

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

@@ -32,6 +32,8 @@
     BackfillCollectionResponse,
     BackfillPostBody,
     BackfillResponse,
+    DryRunBackfillCollectionResponse,
+    DryRunBackfillResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
@@ -206,3 +208,40 @@ 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(
+    body: BackfillPostBody,
+) -> DryRunBackfillCollectionResponse:
+    from_date = timezone.coerce_datetime(body.from_date)
+    to_date = timezone.coerce_datetime(body.to_date)
+
+    try:
+        backfills_dry_run = _create_backfill(
+            dag_id=body.dag_id,
+            from_date=from_date,
+            to_date=to_date,
+            max_active_runs=body.max_active_runs,
+            reverse=body.run_backwards,
+            dag_run_conf=body.dag_run_conf,
+            reprocess_behavior=body.reprocess_behavior,
+            dry_run=True,
+        )
+        backfills = [DryRunBackfillResponse(logical_date=logical_date) for logical_date in backfills_dry_run]
+
+        return DryRunBackfillCollectionResponse(backfills=backfills, total_entries=len(backfills_dry_run))
+
+    except AlreadyRunningBackfill:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="There is already a running backfill for the dag",
+        )