This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 2b68a6293dd Add API endpoint for backfill dag run entries (#67381)
(#70127)
2b68a6293dd is described below
commit 2b68a6293dd9d924edef414afb1e7d227879daf5
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Mon Jul 20 16:42:10 2026 +0200
Add API endpoint for backfill dag run entries (#67381) (#70127)
* Add GET /backfills/{backfill_id}/dag_runs endpoint
Adds a new public API endpoint that returns the BackfillDagRun entries
for a given backfill with joined DagRun state. Users can see what
happened in a backfill: which dates ran, their states (queued, running,
success, failed), and which slots were skipped (with reason).
- BackfillDagRunResponse / BackfillDagRunCollectionResponse models
- LEFT OUTER JOIN via joinedload includes skipped slots (null dag_run_id)
- Pagination via limit/offset, default ordering by sort_ordinal
- 404 when backfill doesn't exist
- 8 unit tests covering happy path, skipped slots, 404, pagination,
empty backfill, and ordering contract
closes: #46250
* Align backfill dag run response identifiers
(cherry picked from commit 6f88f293037cd4934eec4d2317607853cd801e01)
Co-authored-by: Shivam Rastogi <[email protected]>
---
.../api_fastapi/core_api/datamodels/backfills.py | 22 +++
.../core_api/openapi/v2-rest-api-generated.yaml | 148 ++++++++++++++++++
.../core_api/routes/public/backfills.py | 39 +++++
.../src/airflow/ui/openapi-gen/queries/common.ts | 9 ++
.../ui/openapi-gen/queries/ensureQueryData.ts | 17 +++
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 17 +++
.../src/airflow/ui/openapi-gen/queries/queries.ts | 17 +++
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 17 +++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 102 +++++++++++++
.../ui/openapi-gen/requests/services.gen.ts | 34 ++++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 62 ++++++++
.../core_api/routes/public/test_backfills.py | 167 +++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 25 +++
13 files changed, 675 insertions(+), 1 deletion(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
index 52538b37e16..df1ffaca014 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
@@ -24,6 +24,7 @@ from pydantic import AliasPath, Field, NonNegativeInt
from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
from airflow.models.backfill import ReprocessBehavior
+from airflow.utils.state import DagRunState
class BackfillPostBody(StrictBaseModel):
@@ -69,6 +70,27 @@ class BackfillCollectionResponse(BaseModel):
total_entries: int
+class BackfillDagRunResponse(BaseModel):
+ """Serializer for a single BackfillDagRun entry with joined DagRun
state."""
+
+ id: NonNegativeInt
+ backfill_id: NonNegativeInt
+ dag_id: str = Field(validation_alias=AliasPath("backfill", "dag_id"))
+ dag_run_id: str | None = Field(default=None,
validation_alias=AliasPath("dag_run", "run_id"))
+ logical_date: datetime | None
+ partition_key: str | None
+ sort_ordinal: int
+ exception_reason: str | None
+ dag_run_state: DagRunState | None = Field(default=None,
validation_alias=AliasPath("dag_run", "state"))
+
+
+class BackfillDagRunCollectionResponse(BaseModel):
+ """BackfillDagRun Collection serializer for responses."""
+
+ backfill_dag_runs: list[BackfillDagRunResponse]
+ total_entries: int
+
+
class DryRunBackfillResponse(BaseModel):
"""Backfill serializer for responses in dry-run mode."""
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 58ac2e2840c..75caaf4f033 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -1221,6 +1221,85 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ /api/v2/backfills/{backfill_id}/dag_runs:
+ get:
+ tags:
+ - Backfill
+ summary: List Backfill Dag Runs
+ description: List Dag runs associated with a backfill, including skipped
slots.
+ operationId: list_backfill_dag_runs
+ security:
+ - OAuth2PasswordBearer: []
+ - HTTPBearer: []
+ parameters:
+ - name: backfill_id
+ in: path
+ required: true
+ schema:
+ type: integer
+ minimum: 0
+ title: Backfill Id
+ - name: limit
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 0
+ default: 50
+ title: Limit
+ - name: offset
+ in: query
+ required: false
+ schema:
+ type: integer
+ minimum: 0
+ default: 0
+ title: Offset
+ - name: order_by
+ in: query
+ required: false
+ schema:
+ type: array
+ items:
+ type: string
+ description: 'Attributes to order by, multi criteria sort is
supported.
+ Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`'
+ default:
+ - sort_ordinal
+ title: Order By
+ description: 'Attributes to order by, multi criteria sort is
supported. Prefix
+ with `-` for descending order. Supported attributes: `id,
sort_ordinal`'
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/BackfillDagRunCollectionResponse'
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/api/v2/backfills/{backfill_id}/pause:
put:
tags:
@@ -11861,6 +11940,75 @@ components:
- total_entries
title: BackfillCollectionResponse
description: Backfill Collection serializer for responses.
+ BackfillDagRunCollectionResponse:
+ properties:
+ backfill_dag_runs:
+ items:
+ $ref: '#/components/schemas/BackfillDagRunResponse'
+ type: array
+ title: Backfill Dag Runs
+ total_entries:
+ type: integer
+ title: Total Entries
+ type: object
+ required:
+ - backfill_dag_runs
+ - total_entries
+ title: BackfillDagRunCollectionResponse
+ description: BackfillDagRun Collection serializer for responses.
+ BackfillDagRunResponse:
+ properties:
+ id:
+ type: integer
+ minimum: 0.0
+ title: Id
+ backfill_id:
+ type: integer
+ minimum: 0.0
+ title: Backfill Id
+ dag_id:
+ type: string
+ title: Dag Id
+ dag_run_id:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Dag Run Id
+ logical_date:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Logical Date
+ partition_key:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Partition Key
+ sort_ordinal:
+ type: integer
+ title: Sort Ordinal
+ exception_reason:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Exception Reason
+ dag_run_state:
+ anyOf:
+ - $ref: '#/components/schemas/DagRunState'
+ - type: 'null'
+ type: object
+ required:
+ - id
+ - backfill_id
+ - dag_id
+ - logical_date
+ - partition_key
+ - sort_ordinal
+ - exception_reason
+ title: BackfillDagRunResponse
+ description: Serializer for a single BackfillDagRun entry with joined
DagRun
+ state.
BackfillPostBody:
properties:
dag_id:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
index c4971b883a1..62c43616c40 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -35,6 +35,7 @@ from airflow.api_fastapi.common.parameters import QueryLimit,
QueryOffset, SortP
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.datamodels.backfills import (
BackfillCollectionResponse,
+ BackfillDagRunCollectionResponse,
BackfillPostBody,
BackfillResponse,
DryRunBackfillCollectionResponse,
@@ -128,6 +129,44 @@ def get_backfill(
raise HTTPException(status.HTTP_404_NOT_FOUND, "Backfill not found")
+@backfills_router.get(
+ path="/{backfill_id}/dag_runs",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_backfill(method="GET")),
+ ],
+)
+def list_backfill_dag_runs(
+ backfill_id: NonNegativeInt,
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id", "sort_ordinal"],
BackfillDagRun).dynamic_depends(default="sort_ordinal")),
+ ],
+ session: SessionDep,
+) -> BackfillDagRunCollectionResponse:
+ """List Dag runs associated with a backfill, including skipped slots."""
+ backfill = session.get(Backfill, backfill_id)
+ if not backfill:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, f"Backfill with id
{backfill_id} not found")
+
+ select_stmt, total_entries = paginated_select(
+ statement=select(BackfillDagRun)
+ .where(BackfillDagRun.backfill_id == backfill_id)
+ # Load backfill for dag_id; dag_run may be null for skipped slots.
+ .options(joinedload(BackfillDagRun.backfill),
joinedload(BackfillDagRun.dag_run)),
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ return BackfillDagRunCollectionResponse(
+ backfill_dag_runs=list(session.scalars(select_stmt).unique()),
+ total_entries=total_entries,
+ )
+
+
@backfills_router.put(
path="/{backfill_id}/pause",
responses=create_openapi_http_exception_doc(
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index b72cc90eb30..e48d6a8d38c 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -101,6 +101,15 @@ export const useBackfillServiceGetBackfillKey =
"BackfillServiceGetBackfill";
export const UseBackfillServiceGetBackfillKeyFn = ({ backfillId }: {
backfillId: number;
}, queryKey?: Array<unknown>) => [useBackfillServiceGetBackfillKey,
...(queryKey ?? [{ backfillId }])];
+export type BackfillServiceListBackfillDagRunsDefaultResponse =
Awaited<ReturnType<typeof BackfillService.listBackfillDagRuns>>;
+export type BackfillServiceListBackfillDagRunsQueryResult<TData =
BackfillServiceListBackfillDagRunsDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
+export const useBackfillServiceListBackfillDagRunsKey =
"BackfillServiceListBackfillDagRuns";
+export const UseBackfillServiceListBackfillDagRunsKeyFn = ({ backfillId,
limit, offset, orderBy }: {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ orderBy?: string[];
+}, queryKey?: Array<unknown>) => [useBackfillServiceListBackfillDagRunsKey,
...(queryKey ?? [{ backfillId, limit, offset, orderBy }])];
export type BackfillServiceListBackfillsUiDefaultResponse =
Awaited<ReturnType<typeof BackfillService.listBackfillsUi>>;
export type BackfillServiceListBackfillsUiQueryResult<TData =
BackfillServiceListBackfillsUiDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useBackfillServiceListBackfillsUiKey =
"BackfillServiceListBackfillsUi";
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index 74a833b2ccf..4a73de88c2e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -195,6 +195,23 @@ export const ensureUseBackfillServiceGetBackfillData =
(queryClient: QueryClient
backfillId: number;
}) => queryClient.ensureQueryData({ queryKey:
Common.UseBackfillServiceGetBackfillKeyFn({ backfillId }), queryFn: () =>
BackfillService.getBackfill({ backfillId }) });
/**
+* List Backfill Dag Runs
+* List Dag runs associated with a backfill, including skipped slots.
+* @param data The data for the request.
+* @param data.backfillId
+* @param data.limit
+* @param data.offset
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`
+* @returns BackfillDagRunCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseBackfillServiceListBackfillDagRunsData = (queryClient:
QueryClient, { backfillId, limit, offset, orderBy }: {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ orderBy?: string[];
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseBackfillServiceListBackfillDagRunsKeyFn({ backfillId, limit, offset,
orderBy }), queryFn: () => BackfillService.listBackfillDagRuns({ backfillId,
limit, offset, orderBy }) });
+/**
* List Backfills Ui
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index 9db58a9676c..49baa72e3de 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -195,6 +195,23 @@ export const prefetchUseBackfillServiceGetBackfill =
(queryClient: QueryClient,
backfillId: number;
}) => queryClient.prefetchQuery({ queryKey:
Common.UseBackfillServiceGetBackfillKeyFn({ backfillId }), queryFn: () =>
BackfillService.getBackfill({ backfillId }) });
/**
+* List Backfill Dag Runs
+* List Dag runs associated with a backfill, including skipped slots.
+* @param data The data for the request.
+* @param data.backfillId
+* @param data.limit
+* @param data.offset
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`
+* @returns BackfillDagRunCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseBackfillServiceListBackfillDagRuns = (queryClient:
QueryClient, { backfillId, limit, offset, orderBy }: {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ orderBy?: string[];
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseBackfillServiceListBackfillDagRunsKeyFn({ backfillId, limit, offset,
orderBy }), queryFn: () => BackfillService.listBackfillDagRuns({ backfillId,
limit, offset, orderBy }) });
+/**
* List Backfills Ui
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index c10c9635706..440f88ffa64 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -195,6 +195,23 @@ export const useBackfillServiceGetBackfill = <TData =
Common.BackfillServiceGetB
backfillId: number;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseBackfillServiceGetBackfillKeyFn({ backfillId }, queryKey), queryFn:
() => BackfillService.getBackfill({ backfillId }) as TData, ...options });
/**
+* List Backfill Dag Runs
+* List Dag runs associated with a backfill, including skipped slots.
+* @param data The data for the request.
+* @param data.backfillId
+* @param data.limit
+* @param data.offset
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`
+* @returns BackfillDagRunCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useBackfillServiceListBackfillDagRuns = <TData =
Common.BackfillServiceListBackfillDagRunsDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ backfillId, limit, offset,
orderBy }: {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ orderBy?: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseBackfillServiceListBackfillDagRunsKeyFn({ backfillId, limit, offset,
orderBy }, queryKey), queryFn: () => BackfillService.listBackfillDagRuns({
backfillId, limit, offset, orderBy }) as TData, ...options });
+/**
* List Backfills Ui
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index 8c4f6707176..e9e96431829 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -195,6 +195,23 @@ export const useBackfillServiceGetBackfillSuspense =
<TData = Common.BackfillSer
backfillId: number;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseBackfillServiceGetBackfillKeyFn({ backfillId }, queryKey), queryFn:
() => BackfillService.getBackfill({ backfillId }) as TData, ...options });
/**
+* List Backfill Dag Runs
+* List Dag runs associated with a backfill, including skipped slots.
+* @param data The data for the request.
+* @param data.backfillId
+* @param data.limit
+* @param data.offset
+* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`
+* @returns BackfillDagRunCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useBackfillServiceListBackfillDagRunsSuspense = <TData =
Common.BackfillServiceListBackfillDagRunsDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ backfillId, limit, offset,
orderBy }: {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ orderBy?: string[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseBackfillServiceListBackfillDagRunsKeyFn({ backfillId, limit, offset,
orderBy }, queryKey), queryFn: () => BackfillService.listBackfillDagRuns({
backfillId, limit, offset, orderBy }) as TData, ...options });
+/**
* List Backfills Ui
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 16f14472c6a..58b49673193 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -619,6 +619,108 @@ export const $BackfillCollectionResponse = {
description: 'Backfill Collection serializer for responses.'
} as const;
+export const $BackfillDagRunCollectionResponse = {
+ properties: {
+ backfill_dag_runs: {
+ items: {
+ '$ref': '#/components/schemas/BackfillDagRunResponse'
+ },
+ type: 'array',
+ title: 'Backfill Dag Runs'
+ },
+ total_entries: {
+ type: 'integer',
+ title: 'Total Entries'
+ }
+ },
+ type: 'object',
+ required: ['backfill_dag_runs', 'total_entries'],
+ title: 'BackfillDagRunCollectionResponse',
+ description: 'BackfillDagRun Collection serializer for responses.'
+} as const;
+
+export const $BackfillDagRunResponse = {
+ properties: {
+ id: {
+ type: 'integer',
+ minimum: 0,
+ title: 'Id'
+ },
+ backfill_id: {
+ type: 'integer',
+ minimum: 0,
+ title: 'Backfill Id'
+ },
+ dag_id: {
+ type: 'string',
+ title: 'Dag Id'
+ },
+ dag_run_id: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Dag Run Id'
+ },
+ logical_date: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Logical Date'
+ },
+ partition_key: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Partition Key'
+ },
+ sort_ordinal: {
+ type: 'integer',
+ title: 'Sort Ordinal'
+ },
+ exception_reason: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Exception Reason'
+ },
+ dag_run_state: {
+ anyOf: [
+ {
+ '$ref': '#/components/schemas/DagRunState'
+ },
+ {
+ type: 'null'
+ }
+ ]
+ }
+ },
+ type: 'object',
+ required: ['id', 'backfill_id', 'dag_id', 'logical_date', 'partition_key',
'sort_ordinal', 'exception_reason'],
+ title: 'BackfillDagRunResponse',
+ description: 'Serializer for a single BackfillDagRun entry with joined
DagRun state.'
+} as const;
+
export const $BackfillPostBody = {
properties: {
dag_id: {
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 810e9a9a0f5..b9346d8e459 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -3,7 +3,7 @@
import type { CancelablePromise } from './core/CancelablePromise';
import { OpenAPI } from './core/OpenAPI';
import { request as __request } from './core/request';
-import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
+import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
export class AssetService {
/**
@@ -506,6 +506,38 @@ export class BackfillService {
});
}
+ /**
+ * List Backfill Dag Runs
+ * List Dag runs associated with a backfill, including skipped slots.
+ * @param data The data for the request.
+ * @param data.backfillId
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
sort_ordinal`
+ * @returns BackfillDagRunCollectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static listBackfillDagRuns(data: ListBackfillDagRunsData):
CancelablePromise<ListBackfillDagRunsResponse> {
+ return __request(OpenAPI, {
+ method: 'GET',
+ url: '/api/v2/backfills/{backfill_id}/dag_runs',
+ path: {
+ backfill_id: data.backfillId
+ },
+ query: {
+ limit: data.limit,
+ offset: data.offset,
+ order_by: data.orderBy
+ },
+ errors: {
+ 401: 'Unauthorized',
+ 403: 'Forbidden',
+ 404: 'Not Found',
+ 422: 'Validation Error'
+ }
+ });
+ }
+
/**
* Pause Backfill
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 6433c3bfcd6..c9527b1a58d 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -177,6 +177,29 @@ export type BackfillCollectionResponse = {
total_entries: number;
};
+/**
+ * BackfillDagRun Collection serializer for responses.
+ */
+export type BackfillDagRunCollectionResponse = {
+ backfill_dag_runs: Array<BackfillDagRunResponse>;
+ total_entries: number;
+};
+
+/**
+ * Serializer for a single BackfillDagRun entry with joined DagRun state.
+ */
+export type BackfillDagRunResponse = {
+ id: number;
+ backfill_id: number;
+ dag_id: string;
+ dag_run_id?: string | null;
+ logical_date: string | null;
+ partition_key: string | null;
+ sort_ordinal: number;
+ exception_reason: string | null;
+ dag_run_state?: DagRunState | null;
+};
+
/**
* Object used for create backfill request.
*/
@@ -2888,6 +2911,18 @@ export type GetBackfillData = {
export type GetBackfillResponse = BackfillResponse;
+export type ListBackfillDagRunsData = {
+ backfillId: number;
+ limit?: number;
+ offset?: number;
+ /**
+ * Attributes to order by, multi criteria sort is supported. Prefix with
`-` for descending order. Supported attributes: `id, sort_ordinal`
+ */
+ orderBy?: Array<(string)>;
+};
+
+export type ListBackfillDagRunsResponse = BackfillDagRunCollectionResponse;
+
export type PauseBackfillData = {
backfillId: number;
};
@@ -5087,6 +5122,33 @@ export type $OpenApiTs = {
};
};
};
+ '/api/v2/backfills/{backfill_id}/dag_runs': {
+ get: {
+ req: ListBackfillDagRunsData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: BackfillDagRunCollectionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
'/api/v2/backfills/{backfill_id}/pause': {
put: {
req: PauseBackfillData;
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
index d0fdd77e764..a8afb9612d6 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -215,6 +215,173 @@ class TestGetBackfill(TestBackfillEndpoint):
)
+class TestListBackfillDagRuns(TestBackfillEndpoint):
+ def test_list_backfill_dag_runs(self, test_client, session):
+ """Happy path: backfill with mixed dag run states."""
+ (dag,) = self._create_dag_models()
+ from_date = pendulum.parse("2024-01-01")
+ to_date = pendulum.parse("2024-01-03")
+ b = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
+ session.add(b)
+ session.flush()
+
+ dr1 = DagRun(
+ dag_id=dag.dag_id,
+ run_id="backfill__2024-01-01",
+ logical_date=pendulum.parse("2024-01-01"),
+ state=DagRunState.SUCCESS,
+ run_type="scheduled",
+ )
+ dr2 = DagRun(
+ dag_id=dag.dag_id,
+ run_id="backfill__2024-01-02",
+ logical_date=pendulum.parse("2024-01-02"),
+ state=DagRunState.FAILED,
+ run_type="scheduled",
+ )
+ session.add_all([dr1, dr2])
+ session.flush()
+
+ bdr1 = BackfillDagRun(
+ backfill_id=b.id, dag_run_id=dr1.id,
logical_date=pendulum.parse("2024-01-01"), sort_ordinal=1
+ )
+ bdr2 = BackfillDagRun(
+ backfill_id=b.id, dag_run_id=dr2.id,
logical_date=pendulum.parse("2024-01-02"), sort_ordinal=2
+ )
+ session.add_all([bdr1, bdr2])
+ session.commit()
+
+ response = test_client.get(f"/backfills/{b.id}/dag_runs")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["total_entries"] == 2
+ runs = data["backfill_dag_runs"]
+ assert len(runs) == 2
+ assert runs[0]["sort_ordinal"] == 1
+ assert runs[0]["dag_id"] == dag.dag_id
+ assert runs[0]["dag_run_id"] == "backfill__2024-01-01"
+ assert runs[0]["dag_run_state"] == "success"
+ assert runs[1]["sort_ordinal"] == 2
+ assert runs[1]["dag_id"] == dag.dag_id
+ assert runs[1]["dag_run_id"] == "backfill__2024-01-02"
+ assert runs[1]["dag_run_state"] == "failed"
+
+ def test_list_backfill_dag_runs_with_skipped_slots(self, test_client,
session):
+ """Slots skipped due to existing runs have null dag_run_id and
exception_reason set."""
+ (dag,) = self._create_dag_models()
+ b = Backfill(
+ dag_id=dag.dag_id, from_date=pendulum.parse("2024-01-01"),
to_date=pendulum.parse("2024-01-02")
+ )
+ session.add(b)
+ session.flush()
+
+ bdr = BackfillDagRun(
+ backfill_id=b.id,
+ dag_run_id=None,
+ logical_date=pendulum.parse("2024-01-01"),
+ sort_ordinal=1,
+ exception_reason="already exists",
+ )
+ session.add(bdr)
+ session.commit()
+
+ response = test_client.get(f"/backfills/{b.id}/dag_runs")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["total_entries"] == 1
+ run = data["backfill_dag_runs"][0]
+ assert run["dag_id"] == dag.dag_id
+ assert run["dag_run_id"] is None
+ assert run["dag_run_state"] is None
+ assert run["exception_reason"] == "already exists"
+
+ def test_list_backfill_dag_runs_not_found(self, test_client):
+ """Non-existent backfill returns 404."""
+ response = test_client.get("/backfills/999999/dag_runs")
+ assert response.status_code == 404
+
+ def test_list_backfill_dag_runs_pagination(self, test_client, session):
+ """Limit and offset work correctly."""
+ (dag,) = self._create_dag_models()
+ b = Backfill(
+ dag_id=dag.dag_id, from_date=pendulum.parse("2024-01-01"),
to_date=pendulum.parse("2024-01-05")
+ )
+ session.add(b)
+ session.flush()
+
+ for i in range(1, 4):
+ session.add(
+ BackfillDagRun(
+ backfill_id=b.id,
+ dag_run_id=None,
+ logical_date=pendulum.parse(f"2024-01-0{i}"),
+ sort_ordinal=i,
+ exception_reason="already exists",
+ )
+ )
+ session.commit()
+
+ response =
test_client.get(f"/backfills/{b.id}/dag_runs?limit=2&offset=0")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["total_entries"] == 3
+ assert len(data["backfill_dag_runs"]) == 2
+
+ response =
test_client.get(f"/backfills/{b.id}/dag_runs?limit=2&offset=2")
+ assert response.status_code == 200
+ data = response.json()
+ assert len(data["backfill_dag_runs"]) == 1
+
+ def test_list_backfill_dag_runs_empty(self, test_client, session):
+ """Backfill with no dag runs returns empty list."""
+ (dag,) = self._create_dag_models()
+ b = Backfill(
+ dag_id=dag.dag_id, from_date=pendulum.parse("2024-01-01"),
to_date=pendulum.parse("2024-01-02")
+ )
+ session.add(b)
+ session.commit()
+
+ response = test_client.get(f"/backfills/{b.id}/dag_runs")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["total_entries"] == 0
+ assert data["backfill_dag_runs"] == []
+
+ @pytest.mark.parametrize(
+ ("order_by", "expected_first_ordinal"),
+ [
+ ("sort_ordinal", 1),
+ ("-sort_ordinal", 3),
+ ("id", 1),
+ ],
+ )
+ def test_list_backfill_dag_runs_ordering(self, order_by,
expected_first_ordinal, test_client, session):
+ """Verify sort contract for allowed order_by values."""
+ (dag,) = self._create_dag_models()
+ b = Backfill(
+ dag_id=dag.dag_id, from_date=pendulum.parse("2024-01-01"),
to_date=pendulum.parse("2024-01-03")
+ )
+ session.add(b)
+ session.flush()
+
+ for i in range(1, 4):
+ session.add(
+ BackfillDagRun(
+ backfill_id=b.id,
+ dag_run_id=None,
+ logical_date=pendulum.parse(f"2024-01-0{i}"),
+ sort_ordinal=i,
+ exception_reason="already exists",
+ )
+ )
+ session.commit()
+
+ response =
test_client.get(f"/backfills/{b.id}/dag_runs?order_by={order_by}")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["backfill_dag_runs"][0]["sort_ordinal"] ==
expected_first_ordinal
+
+
class TestCreateBackfill(TestBackfillEndpoint):
@pytest.mark.parametrize(
("repro_act", "repro_exp"),
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 8804d2040ea..83bd18eb942 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -1365,6 +1365,22 @@ class AssetStateStoreResponse(BaseModel):
last_updated_by: AssetStateStoreLastUpdatedBy | None = None
+class BackfillDagRunResponse(BaseModel):
+ """
+ Serializer for a single BackfillDagRun entry with joined DagRun state.
+ """
+
+ id: Annotated[int, Field(ge=0, title="Id")]
+ backfill_id: Annotated[int, Field(ge=0, title="Backfill Id")]
+ dag_id: Annotated[str, Field(title="Dag Id")]
+ dag_run_id: Annotated[str | None, Field(title="Dag Run Id")] = None
+ logical_date: Annotated[datetime | None, Field(title="Logical Date")] =
None
+ partition_key: Annotated[str | None, Field(title="Partition Key")] = None
+ sort_ordinal: Annotated[int, Field(title="Sort Ordinal")]
+ exception_reason: Annotated[str | None, Field(title="Exception Reason")] =
None
+ dag_run_state: DagRunState | None = None
+
+
class BackfillPostBody(BaseModel):
"""
Object used for create backfill request.
@@ -2251,6 +2267,15 @@ class BackfillCollectionResponse(BaseModel):
total_entries: Annotated[int, Field(title="Total Entries")]
+class BackfillDagRunCollectionResponse(BaseModel):
+ """
+ BackfillDagRun Collection serializer for responses.
+ """
+
+ backfill_dag_runs: Annotated[list[BackfillDagRunResponse],
Field(title="Backfill Dag Runs")]
+ total_entries: Annotated[int, Field(title="Total Entries")]
+
+
class BulkBodyConnectionBody(BaseModel):
model_config = ConfigDict(
extra="forbid",
