Use NonNegativeInt for backfill_id (#49691) (#49716)

* use NonNegativeInt for backfill_id

* fix aws failing test

* update BackfillResponse
(cherry picked from commit e564f1b)

Co-authored-by: Kalyan R <[email protected]>

Author: 2 people

airflow-core/src/airflow/api_fastapi/auth/managers/models/resource_details.py

@@ -20,6 +20,8 @@
 from dataclasses import dataclass
 from enum import Enum
 
+from pydantic import NonNegativeInt
+
 
 @dataclass
 class ConfigurationDetails:
@@ -46,7 +48,7 @@ class DagDetails:
 class BackfillDetails:
     """Represents the details of a backfill."""
 
-    id: str | None = None
+    id: NonNegativeInt | None = None
 
 
 @dataclass

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

@@ -19,6 +19,8 @@
 
 from datetime import datetime
 
+from pydantic import NonNegativeInt
+
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
 from airflow.models.backfill import ReprocessBehavior
 
@@ -38,7 +40,7 @@ class BackfillPostBody(StrictBaseModel):
 class BackfillResponse(BaseModel):
     """Base serializer for Backfill."""
 
-    id: int
+    id: NonNegativeInt
     dag_id: str
     from_date: datetime
     to_date: datetime

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

@@ -595,6 +595,7 @@ components:
       properties:
         id:
           type: integer
+          minimum: 0.0
           title: Id
         dag_id:
           type: string

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

@@ -951,7 +951,8 @@ paths:
         in: path
         required: true
         schema:
-          type: string
+          type: integer
+          minimum: 0
           title: Backfill Id
       responses:
         '200':
@@ -997,6 +998,8 @@ paths:
         in: path
         required: true
         schema:
+          type: integer
+          minimum: 0
           title: Backfill Id
       responses:
         '200':
@@ -1048,6 +1051,8 @@ paths:
         in: path
         required: true
         schema:
+          type: integer
+          minimum: 0
           title: Backfill Id
       responses:
         '200':
@@ -1099,6 +1104,8 @@ paths:
         in: path
         required: true
         schema:
+          type: integer
+          minimum: 0
           title: Backfill Id
       responses:
         '200':
@@ -7039,6 +7046,7 @@ components:
       properties:
         id:
           type: integer
+          minimum: 0.0
           title: Id
         dag_id:
           type: string

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

@@ -20,6 +20,7 @@
 
 from fastapi import Depends, HTTPException, status
 from fastapi.exceptions import RequestValidationError
+from pydantic import NonNegativeInt
 from sqlalchemy import select, update
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
@@ -98,7 +99,7 @@ def list_backfills(
     ],
 )
 def get_backfill(
-    backfill_id: str,
+    backfill_id: NonNegativeInt,
     session: SessionDep,
 ) -> BackfillResponse:
     backfill = session.get(Backfill, backfill_id)
@@ -121,7 +122,7 @@ def get_backfill(
         Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.RUN)),
     ],
 )
-def pause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
+def pause_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> BackfillResponse:
     b = session.get(Backfill, backfill_id)
     if not b:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Could not find backfill with id {backfill_id}")
@@ -147,7 +148,7 @@ def pause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
         Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.RUN)),
     ],
 )
-def unpause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
+def unpause_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> BackfillResponse:
     b = session.get(Backfill, backfill_id)
     if not b:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Could not find backfill with id {backfill_id}")
@@ -172,7 +173,7 @@ def unpause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
         Depends(requires_access_backfill(method="PUT")),
     ],
 )
-def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
+def cancel_backfill(backfill_id: NonNegativeInt, session: SessionDep) -> BackfillResponse:
     b: Backfill = session.get(Backfill, backfill_id)
     if not b:
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"Could not find backfill with id {backfill_id}")

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -23,6 +23,7 @@
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import OAuth2PasswordBearer
 from jwt import ExpiredSignatureError, InvalidTokenError
+from pydantic import NonNegativeInt
 
 from airflow.api_fastapi.app import get_auth_manager
 from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
@@ -198,7 +199,7 @@ def inner(
         request: Request,
         user: Annotated[BaseUser | None, Depends(get_user)] = None,
     ) -> None:
-        backfill_id: str | None = request.path_params.get("backfill_id")
+        backfill_id: NonNegativeInt | None = request.path_params.get("backfill_id")
 
         _requires_access(
             is_authorized_callback=lambda: get_auth_manager().is_authorized_backfill(

airflow-core/src/airflow/models/dag.py

@@ -116,6 +116,7 @@
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
 if TYPE_CHECKING:
+    from pydantic import NonNegativeInt
     from sqlalchemy.orm.query import Query
     from sqlalchemy.orm.session import Session
 
@@ -257,7 +258,7 @@ def _create_orm_dagrun(
     state: DagRunState | None,
     run_type: DagRunType,
     creating_job_id: int | None,
-    backfill_id: int | None,
+    backfill_id: NonNegativeInt | None,
     triggered_by: DagRunTriggeredByType,
     session: Session = NEW_SESSION,
 ) -> DagRun:
@@ -1795,7 +1796,7 @@ def create_dagrun(
         state: DagRunState,
         start_date: datetime | None = None,
         creating_job_id: int | None = None,
-        backfill_id: int | None = None,
+        backfill_id: NonNegativeInt | None = None,
         session: Session = NEW_SESSION,
     ) -> DagRun:
         """

airflow-core/src/airflow/models/dagrun.py

@@ -93,6 +93,7 @@
     from datetime import datetime
 
     from opentelemetry.sdk.trace import Span
+    from pydantic import NonNegativeInt
     from sqlalchemy.orm import Query, Session
 
     from airflow.models.baseoperator import BaseOperator
@@ -290,7 +291,7 @@ def __init__(
         creating_job_id: int | None = None,
         data_interval: tuple[datetime, datetime] | None = None,
         triggered_by: DagRunTriggeredByType | None = None,
-        backfill_id: int | None = None,
+        backfill_id: NonNegativeInt | None = None,
         bundle_version: str | None = None,
     ):
         # For manual runs where logical_date is None, ensure no data_interval is set.

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

@@ -270,7 +270,7 @@ export const UseBackfillServiceGetBackfillKeyFn = (
   {
     backfillId,
   }: {
-    backfillId: string;
+    backfillId: number;
   },
   queryKey?: Array<unknown>,
 ) => [useBackfillServiceGetBackfillKey, ...(queryKey ?? [{ backfillId }])];

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

@@ -354,7 +354,7 @@ export const ensureUseBackfillServiceGetBackfillData = (
   {
     backfillId,
   }: {
-    backfillId: string;
+    backfillId: number;
   },
 ) =>
   queryClient.ensureQueryData({