AIP-84 Handle backfill for DAGs with when depends_on_past is True

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

@@ -1285,11 +1285,11 @@ paths:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
           description: Conflict
         '422':
-          description: Validation Error
           content:
             application/json:
               schema:
-                $ref: '#/components/schemas/HTTPValidationError'
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Unprocessable Entity
   /public/backfills/{backfill_id}:
     get:
       tags:
@@ -1525,11 +1525,11 @@ paths:
               schema:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
         '422':
-          description: Validation Error
+          description: Unprocessable Entity
           content:
             application/json:
               schema:
-                $ref: '#/components/schemas/HTTPValidationError'
+                $ref: '#/components/schemas/HTTPExceptionResponse'
   /public/connections/{connection_id}:
     delete:
       tags:

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

@@ -45,6 +45,8 @@
     Backfill,
     BackfillDagRun,
     DagNoScheduleException,
+    InvalidBackfillDirection,
+    InvalidReprocessBehavior,
     _create_backfill,
     _do_dry_run,
 )
@@ -184,10 +186,7 @@ def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
 @backfills_router.post(
     path="",
     responses=create_openapi_http_exception_doc(
-        [
-            status.HTTP_404_NOT_FOUND,
-            status.HTTP_409_CONFLICT,
-        ]
+        [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
     ),
 )
 def create_backfill(
@@ -213,9 +212,21 @@ def create_backfill(
         )
     except DagNoScheduleException:
         raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail=f"{backfill_request.dag_id} has no schedule",
         )
+    except InvalidReprocessBehavior:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+            "You must set reprocess behavior to reprocess completed or reprocess failed.",
+        )
+
+    except InvalidBackfillDirection:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+        )
 
     except DagNotFound:
         raise HTTPException(
@@ -227,10 +238,7 @@ def create_backfill(
 @backfills_router.post(
     path="/dry_run",
     responses=create_openapi_http_exception_doc(
-        [
-            status.HTTP_404_NOT_FOUND,
-            status.HTTP_409_CONFLICT,
-        ]
+        [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
     ),
 )
 def create_backfill_dry_run(
@@ -259,6 +267,18 @@ def create_backfill_dry_run(
         )
     except DagNoScheduleException:
         raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail=f"{body.dag_id} has no schedule",
         )
+
+    except InvalidReprocessBehavior:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"{body.dag_id} has tasks for which depends_on_past=True. "
+            "You must set reprocess behavior to reprocess completed or reprocess failed.",
+        )
+    except InvalidBackfillDirection:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+        )

airflow/models/backfill.py

@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
     """
 
 
+class InvalidBackfillDirection(AirflowException):
+    """
+    Raised when backfill is attempted in reverse order with tasks that depend on past runs.
+
+    :meta private:
+    """
+
+
+class InvalidReprocessBehavior(AirflowException):
+    """
+    Raised when reprocess behavior is not set for tasks with depends_on_past=True.
+
+    :meta private:
+    """
+
+
 class ReprocessBehavior(str, Enum):
     """
     Internal enum for setting reprocess behavior in a backfill.
@@ -188,18 +204,16 @@ def _get_dag_run_no_create_reason(dr, reprocess_behavior: ReprocessBehavior) ->
 
 
 def _validate_backfill_params(dag, reverse, reprocess_behavior: ReprocessBehavior | None):
-    depends_on_past = None
     depends_on_past = any(x.depends_on_past for x in dag.tasks)
     if depends_on_past:
         if reverse is True:
-            raise ValueError(
-                "Backfill cannot be run in reverse when the dag has tasks where depends_on_past=True"
+            raise InvalidBackfillDirection(
+                "Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True"
             )
         if reprocess_behavior in (None, ReprocessBehavior.NONE):
-            raise ValueError(
-                "Dag has task for which depends_on_past is true. "
-                "You must set reprocess behavior to reprocess completed or "
-                "reprocess failed"
+            raise InvalidReprocessBehavior(
+                "DAG has tasks for which depends_on_past=True. "
+                "You must set reprocess behavior to reprocess completed or reprocess failed."
             )
 
 

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

@@ -830,7 +830,7 @@ export class BackfillService {
         403: "Forbidden",
         404: "Not Found",
         409: "Conflict",
-        422: "Validation Error",
+        422: "Unprocessable Entity",
       },
     });
   }
@@ -950,7 +950,7 @@ export class BackfillService {
         403: "Forbidden",
         404: "Not Found",
         409: "Conflict",
-        422: "Validation Error",
+        422: "Unprocessable Entity",
       },
     });
   }

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

@@ -2792,9 +2792,9 @@ export type $OpenApiTs = {
          */
         409: HTTPExceptionResponse;
         /**
-         * Validation Error
+         * Unprocessable Entity
          */
-        422: HTTPValidationError;
+        422: HTTPExceptionResponse;
       };
     };
   };
@@ -2943,9 +2943,9 @@ export type $OpenApiTs = {
          */
         409: HTTPExceptionResponse;
         /**
-         * Validation Error
+         * Unprocessable Entity
          */
-        422: HTTPValidationError;
+        422: HTTPExceptionResponse;
       };
     };
   };