This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c57abb98ec6 Add cursor based pagination for get_dag_runs endpoint
(#65604)
c57abb98ec6 is described below
commit c57abb98ec62ff0d59e5c78decb67e94cf57480c
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Thu Apr 23 22:23:23 2026 +0200
Add cursor based pagination for get_dag_runs endpoint (#65604)
* Add cursor based pagination for get_dag_runs endpoint
Add cursor-based (keyset) pagination as an alternative to offset-based
pagination on the get_dag_runs endpoint. Offset pagination remains the
default and is not deprecated. Mirrors the pattern introduced for
get_task_instances in #64845.
- Response schema adds next_cursor / previous_cursor and makes
total_entries optional (populated only for offset pagination).
- Fetch limit+1 rows to detect last page and return next_cursor=null
when no more results exist.
- Rewire the DagRuns listing page to use cursor pagination via the
DataTable's cursorPagination prop, which eliminates the COUNT(*)
query for large datasets.
* Small Adjustments
* Small Adjustments
---
.../src/airflow/api_fastapi/common/cursors.py | 2 +-
.../src/airflow/api_fastapi/common/parameters.py | 31 ++++-
.../api_fastapi/core_api/datamodels/dag_run.py | 27 ++++-
.../core_api/openapi/v2-rest-api-generated.yaml | 71 ++++++++++-
.../api_fastapi/core_api/routes/public/dag_run.py | 102 ++++++++++++----
.../core_api/routes/public/task_instances.py | 2 +-
.../src/airflow/ui/openapi-gen/queries/common.ts | 5 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 15 ++-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 15 ++-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 15 ++-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 15 ++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 47 +++++++-
.../ui/openapi-gen/requests/services.gen.ts | 11 ++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 26 ++++-
airflow-core/src/airflow/ui/src/pages/DagRuns.tsx | 12 +-
.../ui/src/pages/TaskInstances/TaskInstances.tsx | 4 +-
.../unit/api_fastapi/common/test_parameters.py | 35 ++++++
.../core_api/routes/public/test_dag_run.py | 130 +++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 31 ++++-
19 files changed, 543 insertions(+), 53 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/cursors.py
b/airflow-core/src/airflow/api_fastapi/common/cursors.py
index 0d06aa10b2e..21bfa23d4a4 100644
--- a/airflow-core/src/airflow/api_fastapi/common/cursors.py
+++ b/airflow-core/src/airflow/api_fastapi/common/cursors.py
@@ -103,7 +103,7 @@ def encode_cursor(row: Any, sort_param: SortParam) -> str:
if not resolved:
raise ValueError("SortParam has no resolved columns.")
- parts = [getattr(row, attr_name, None) for attr_name, _col, _desc in
resolved]
+ parts = [sort_param.row_value(row, attr_name) for attr_name, _col, _desc
in resolved]
payload = msgspec.msgpack.encode(parts)
return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py
b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index 28180b4d81f..98963d50a4e 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -308,6 +308,9 @@ class SortParam(BaseParam[list[str]]):
resolved: list[tuple[str, ColumnElement, bool]] = []
for order_by_value in order_by_values:
lstriped_orderby = order_by_value.lstrip("-")
+ # Store the user-facing name in the resolved tuple. ``row_value``
resolves
+ # it back to the actual row accessor via ``to_replace`` when
reading values
+ # for cursor encoding.
attr_name = lstriped_orderby
column: Column | None = None
if self.to_replace:
@@ -330,7 +333,8 @@ class SortParam(BaseParam[list[str]]):
primary_key_column = self.get_primary_key_column()
pk_name = self.get_primary_key_string()
- if not any(name == pk_name for name, _, _ in resolved):
+ resolved_column_keys = {getattr(col, "key", None) for _, col, _ in
resolved}
+ if pk_name not in resolved_column_keys:
pk_desc = bool(order_by_values and
order_by_values[0].startswith("-"))
resolved.append((pk_name, primary_key_column, pk_desc))
@@ -352,6 +356,31 @@ class SortParam(BaseParam[list[str]]):
"""Return resolved sort columns as (attr_name, column_element,
is_descending) tuples."""
return self._resolve()
+ def row_value(self, row: Any, name: str) -> Any:
+ """
+ Extract the sort-key value for ``name`` from a result row.
+
+ Resolves the accessor through ``to_replace`` for string aliases
+ (e.g. ``{"dag_run_id": "run_id"}``); otherwise reads ``name`` directly.
+ """
+ if self.to_replace:
+ replacement = self.to_replace.get(name)
+ if isinstance(replacement, str):
+ return getattr(row, replacement, None)
+ if replacement is not None:
+ # TODO: Column-form ``to_replace`` (e.g. ``{"last_run_state":
DagRun.state}``)
+ # isn't supported for cursor pagination — no endpoint that
uses cursor
+ # pagination needs it today. When one does, decide how the row
exposes the
+ # value (projected label on the SELECT, eagerly loaded
relationship, etc.)
+ # and wire it up here. Raising loudly so a future caller
doesn't silently
+ # get ``None`` cursor tokens.
+ raise NotImplementedError(
+ f"Cursor pagination does not support column-form
``to_replace`` mapping for "
+ f"``{name}``. Use a string alias in ``to_replace`` or sort
by a primary-model "
+ f"attribute."
+ )
+ return getattr(row, name, None)
+
def get_primary_key_column(self) -> Column:
"""Get the primary key column of the model of SortParam object."""
return inspect(self.model).primary_key[0]
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
index 334917160b5..8e5f31e09fa 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
@@ -100,10 +100,33 @@ class DAGRunResponse(BaseModel):
class DAGRunCollectionResponse(BaseModel):
- """DAG Run Collection serializer for responses."""
+ """
+ DAG Run collection response supporting both offset and cursor pagination.
+
+ A single flat model is used instead of a discriminated union
+ (``Annotated[Offset | Cursor, Field(discriminator=...)]``) because
+ the OpenAPI ``oneOf`` + ``discriminator`` construct is not handled
+ correctly by ``@hey-api/openapi-ts`` /
``@7nohe/openapi-react-query-codegen``:
+ return types degrade to ``unknown`` in JSDoc and can produce
+ incorrect TypeScript types (see hey-api/openapi-ts#1613, #3270).
+ """
dag_runs: Iterable[DAGRunResponse]
- total_entries: int
+ total_entries: int | None = Field(
+ default=None,
+ description="Total number of matching items. Populated for offset
pagination, "
+ "``null`` when using cursor pagination.",
+ )
+ next_cursor: str | None = Field(
+ default=None,
+ description="Token pointing to the next page. Populated for cursor
pagination, "
+ "``null`` when using offset pagination or when there is no next page.",
+ )
+ previous_cursor: str | None = Field(
+ default=None,
+ description="Token pointing to the previous page. Populated for cursor
pagination, "
+ "``null`` when using offset pagination or when on the first page.",
+ )
class TriggerDAGRunPostBody(StrictBaseModel):
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 6e01355d9fa..2c089c329f7 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
@@ -2023,7 +2023,25 @@ paths:
This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs
for
- all DAGs.'
+ all DAGs.
+
+
+ Supports two pagination modes:
+
+
+ **Offset (default):** use `limit` and `offset` query parameters.
Returns `total_entries`.
+
+
+ **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor`
+ from the response).
+
+ When `cursor` is provided, `offset` is ignored and `total_entries` is
not
+ returned.
+
+ ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor``
+ is ``null``
+
+ on the first page.'
operationId: get_dag_runs
security:
- OAuth2PasswordBearer: []
@@ -2035,6 +2053,20 @@ paths:
schema:
type: string
title: Dag Id
+ - name: cursor
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ description: Cursor for keyset-based pagination. Pass an empty
string for
+ the first page, then use ``next_cursor`` from the response. When
``cursor``
+ is provided, ``offset`` is ignored.
+ title: Cursor
+ description: Cursor for keyset-based pagination. Pass an empty string
for
+ the first page, then use ``next_cursor`` from the response. When
``cursor``
+ is provided, ``offset`` is ignored.
- name: limit
in: query
required: false
@@ -10752,14 +10784,45 @@ components:
type: array
title: Dag Runs
total_entries:
- type: integer
+ anyOf:
+ - type: integer
+ - type: 'null'
title: Total Entries
+ description: Total number of matching items. Populated for offset
pagination,
+ ``null`` when using cursor pagination.
+ next_cursor:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Next Cursor
+ description: Token pointing to the next page. Populated for cursor
pagination,
+ ``null`` when using offset pagination or when there is no next
page.
+ previous_cursor:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Previous Cursor
+ description: Token pointing to the previous page. Populated for
cursor pagination,
+ ``null`` when using offset pagination or when on the first page.
type: object
required:
- dag_runs
- - total_entries
title: DAGRunCollectionResponse
- description: DAG Run Collection serializer for responses.
+ description: 'DAG Run collection response supporting both offset and
cursor
+ pagination.
+
+
+ A single flat model is used instead of a discriminated union
+
+ (``Annotated[Offset | Cursor, Field(discriminator=...)]``) because
+
+ the OpenAPI ``oneOf`` + ``discriminator`` construct is not handled
+
+ correctly by ``@hey-api/openapi-ts`` /
``@7nohe/openapi-react-query-codegen``:
+
+ return types degrade to ``unknown`` in JSDoc and can produce
+
+ incorrect TypeScript types (see hey-api/openapi-ts#1613, #3270).'
DAGRunPatchBody:
properties:
state:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index b1f86aed1e8..88e4ec564f0 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -35,8 +35,14 @@ from airflow.api.common.mark_tasks import (
)
from airflow.api_fastapi.app import get_auth_manager
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity, DagDetails
+from airflow.api_fastapi.common.cursors import (
+ apply_cursor_filter,
+ encode_cursor,
+ make_backward_cursor,
+ parse_cursor,
+)
from airflow.api_fastapi.common.dagbag import DagBagDep, get_dag_for_run,
get_latest_version_of_dag
-from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.db.common import SessionDep,
apply_filters_to_select, paginated_select
from airflow.api_fastapi.common.db.dag_runs import (
attach_dag_versions_to_runs,
eager_load_dag_run_for_list,
@@ -65,6 +71,7 @@ from airflow.api_fastapi.common.parameters import (
)
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.common.types import Mimetype
+from airflow.api_fastapi.core_api.base import OrmClause
from airflow.api_fastapi.core_api.datamodels.assets import
AssetEventCollectionResponse
from airflow.api_fastapi.core_api.datamodels.dag_run import (
DAGRunClearBody,
@@ -417,12 +424,28 @@ def get_dag_runs(
dag_id_pattern: Annotated[_SearchParam,
Depends(search_param_factory(DagRun.dag_id, "dag_id_pattern"))],
partition_key_pattern: QueryDagRunPartitionKeySearch,
consuming_asset_pattern: QueryConsumingAssetPatternSearch,
+ cursor: str | None = Query(
+ None,
+ description="Cursor for keyset-based pagination. "
+ "Pass an empty string for the first page, then use ``next_cursor``
from the response. "
+ "When ``cursor`` is provided, ``offset`` is ignored.",
+ ),
) -> DAGRunCollectionResponse:
"""
Get all DAG Runs.
This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs for
all DAGs.
+
+ Supports two pagination modes:
+
+ **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
+
+ **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+ When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+ ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+ on the first page.
"""
+ use_cursor = cursor is not None
query = select(DagRun).options(*eager_load_dag_run_for_list())
if dag_id != "~":
@@ -433,27 +456,66 @@ def get_dag_runs(
if dag_version.value:
query = query.join(DagVersion, DagRun.created_dag_version_id ==
DagVersion.id)
+ filters: list[OrmClause] = [
+ run_after,
+ logical_date,
+ start_date_range,
+ end_date_range,
+ update_at_range,
+ duration_range,
+ conf_contains,
+ state,
+ run_type,
+ dag_version,
+ bundle_version,
+ readable_dag_runs_filter,
+ run_id_pattern,
+ triggering_user_name_pattern,
+ dag_id_pattern,
+ partition_key_pattern,
+ consuming_asset_pattern,
+ ]
+
+ if use_cursor:
+ # Fetch one extra row so we can detect whether a next page exists.
+ page_limit = cast(
+ "int", limit.value
+ ) # LimitFilter value is guaranteed to be set to the default value of
QueryLimit
+ cursor_limit = LimitFilter().set_value(page_limit + 1)
+ dag_run_select = apply_filters_to_select(statement=query,
filters=[*filters, order_by, cursor_limit])
+
+ is_backward = False
+ if cursor:
+ token, is_backward = parse_cursor(cursor)
+ if is_backward:
+ dag_run_select = order_by.to_orm(dag_run_select, reversed=True)
+ dag_run_select = apply_cursor_filter(dag_run_select, token,
order_by, is_backward=is_backward)
+
+ fetched = list(session.scalars(dag_run_select).unique())
+ has_more = len(fetched) > page_limit
+ dag_runs = fetched[:page_limit]
+
+ if is_backward:
+ dag_runs.reverse()
+ has_prev = has_more
+ has_next = True
+ else:
+ has_prev = bool(cursor)
+ has_next = has_more
+
+ attach_dag_versions_to_runs(dag_runs, session=session)
+
+ return DAGRunCollectionResponse(
+ dag_runs=dag_runs,
+ next_cursor=(encode_cursor(dag_runs[-1], order_by) if has_next and
dag_runs else None),
+ previous_cursor=(
+ make_backward_cursor(encode_cursor(dag_runs[0], order_by)) if
has_prev and dag_runs else None
+ ),
+ )
+
dag_run_select, total_entries = paginated_select(
statement=query,
- filters=[
- run_after,
- logical_date,
- start_date_range,
- end_date_range,
- update_at_range,
- duration_range,
- conf_contains,
- state,
- run_type,
- dag_version,
- bundle_version,
- readable_dag_runs_filter,
- run_id_pattern,
- triggering_user_name_pattern,
- dag_id_pattern,
- partition_key_pattern,
- consuming_asset_pattern,
- ],
+ filters=filters,
order_by=order_by,
offset=offset,
limit=limit,
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
index 3d3fa538b5a..a6ed8f53aef 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
@@ -551,7 +551,7 @@ def get_task_instances(
# Fetch one extra row so we can detect whether a next page exists.
page_limit = cast(
"int", limit.value
- ) # LimitFilter value is guaranteed to be set of the default value of
QueryLimit
+ ) # LimitFilter value is guaranteed to be set to the default value of
QueryLimit
cursor_limit = LimitFilter().set_value(page_limit + 1)
task_instance_select = apply_filters_to_select(
statement=query, filters=[*filters, order_by, cursor_limit]
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 33f255bb9a9..e7b2bdd6318 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -143,10 +143,11 @@ export const UseDagRunServiceGetUpstreamAssetEventsKeyFn
= ({ dagId, dagRunId }:
export type DagRunServiceGetDagRunsDefaultResponse = Awaited<ReturnType<typeof
DagRunService.getDagRuns>>;
export type DagRunServiceGetDagRunsQueryResult<TData =
DagRunServiceGetDagRunsDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useDagRunServiceGetDagRunsKey = "DagRunServiceGetDagRuns";
-export const UseDagRunServiceGetDagRunsKeyFn = ({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDateLt, startDateLte, state, triggeringUserNamePa [...]
+export const UseDagRunServiceGetDagRunsKeyFn = ({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDateLt, startDateLte, state, triggeringUs [...]
bundleVersion?: string;
confContains?: string;
consumingAssetPattern?: string;
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -182,7 +183,7 @@ export const UseDagRunServiceGetDagRunsKeyFn = ({
bundleVersion, confContains, c
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: Array<unknown>) => [useDagRunServiceGetDagRunsKey, ...(queryKey
?? [{ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern,
dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt,
endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte,
logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern,
runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType,
startDateGt, startDateGte, startDateLt, startDa [...]
+}, queryKey?: Array<unknown>) => [useDagRunServiceGetDagRunsKey, ...(queryKey
?? [{ bundleVersion, confContains, consumingAssetPattern, cursor, dagId,
dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte,
endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy,
partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte,
runIdPattern, runType, startDateGt, startDateGte, startDateLt, [...]
export type DagRunServiceWaitDagRunUntilFinishedDefaultResponse =
Awaited<ReturnType<typeof DagRunService.waitDagRunUntilFinished>>;
export type DagRunServiceWaitDagRunUntilFinishedQueryResult<TData =
DagRunServiceWaitDagRunUntilFinishedDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useDagRunServiceWaitDagRunUntilFinishedKey =
"DagRunServiceWaitDagRunUntilFinished";
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 e9d82da6aac..68bb6e8ded8 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -261,8 +261,18 @@ export const
ensureUseDagRunServiceGetUpstreamAssetEventsData = (queryClient: Qu
* Get all DAG Runs.
*
* This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs for
all DAGs.
+*
+* Supports two pagination modes:
+*
+* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
+*
+* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+* on the first page.
* @param data The data for the request.
* @param data.dagId
+* @param data.cursor Cursor for keyset-based pagination. Pass an empty string
for the first page, then use ``next_cursor`` from the response. When ``cursor``
is provided, ``offset`` is ignored.
* @param data.limit
* @param data.offset
* @param data.runAfterGte
@@ -303,10 +313,11 @@ export const
ensureUseDagRunServiceGetUpstreamAssetEventsData = (queryClient: Qu
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseDagRunServiceGetDagRunsData = (queryClient: QueryClient,
{ bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern,
dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt,
endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte,
logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern,
runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType,
startDateGt, startDateGte, startDateLt, startDateL [...]
+export const ensureUseDagRunServiceGetDagRunsData = (queryClient: QueryClient,
{ bundleVersion, confContains, consumingAssetPattern, cursor, dagId,
dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte,
endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy,
partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte,
runIdPattern, runType, startDateGt, startDateGte, startDateLt, st [...]
bundleVersion?: string;
confContains?: string;
consumingAssetPattern?: string;
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -342,7 +353,7 @@ export const ensureUseDagRunServiceGetDagRunsData =
(queryClient: QueryClient, {
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDateLt, star [...]
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDate [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if
requested.
* 🚧 This is an experimental endpoint and may change or be removed without
notice.Successful response are streamed as newline-delimited JSON (NDJSON).
Each line is a JSON object representing the DAG run state.
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 af819e0110e..cf69e71c4f5 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -261,8 +261,18 @@ export const
prefetchUseDagRunServiceGetUpstreamAssetEvents = (queryClient: Quer
* Get all DAG Runs.
*
* This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs for
all DAGs.
+*
+* Supports two pagination modes:
+*
+* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
+*
+* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+* on the first page.
* @param data The data for the request.
* @param data.dagId
+* @param data.cursor Cursor for keyset-based pagination. Pass an empty string
for the first page, then use ``next_cursor`` from the response. When ``cursor``
is provided, ``offset`` is ignored.
* @param data.limit
* @param data.offset
* @param data.runAfterGte
@@ -303,10 +313,11 @@ export const
prefetchUseDagRunServiceGetUpstreamAssetEvents = (queryClient: Quer
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseDagRunServiceGetDagRuns = (queryClient: QueryClient, {
bundleVersion, confContains, consumingAssetPattern, dagId, dagIdPattern,
dagVersion, durationGt, durationGte, durationLt, durationLte, endDateGt,
endDateGte, endDateLt, endDateLte, limit, logicalDateGt, logicalDateGte,
logicalDateLt, logicalDateLte, offset, orderBy, partitionKeyPattern,
runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, runType,
startDateGt, startDateGte, startDateLt, startDateLte [...]
+export const prefetchUseDagRunServiceGetDagRuns = (queryClient: QueryClient, {
bundleVersion, confContains, consumingAssetPattern, cursor, dagId,
dagIdPattern, dagVersion, durationGt, durationGte, durationLt, durationLte,
endDateGt, endDateGte, endDateLt, endDateLte, limit, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, offset, orderBy,
partitionKeyPattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte,
runIdPattern, runType, startDateGt, startDateGte, startDateLt, star [...]
bundleVersion?: string;
confContains?: string;
consumingAssetPattern?: string;
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -342,7 +353,7 @@ export const prefetchUseDagRunServiceGetDagRuns =
(queryClient: QueryClient, { b
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDateLt, startD [...]
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, runAfterLte, runIdPattern, runType, startDateGt, startDateGte,
startDateLt [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if
requested.
* 🚧 This is an experimental endpoint and may change or be removed without
notice.Successful response are streamed as newline-delimited JSON (NDJSON).
Each line is a JSON object representing the DAG run state.
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 87d27071007..efd94101d23 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -261,8 +261,18 @@ export const useDagRunServiceGetUpstreamAssetEvents =
<TData = Common.DagRunServ
* Get all DAG Runs.
*
* This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs for
all DAGs.
+*
+* Supports two pagination modes:
+*
+* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
+*
+* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+* on the first page.
* @param data The data for the request.
* @param data.dagId
+* @param data.cursor Cursor for keyset-based pagination. Pass an empty string
for the first page, then use ``next_cursor`` from the response. When ``cursor``
is provided, ``offset`` is ignored.
* @param data.limit
* @param data.offset
* @param data.runAfterGte
@@ -303,10 +313,11 @@ export const useDagRunServiceGetUpstreamAssetEvents =
<TData = Common.DagRunServ
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagRunServiceGetDagRuns = <TData =
Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfterLt, run [...]
+export const useDagRunServiceGetDagRuns = <TData =
Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfte [...]
bundleVersion?: string;
confContains?: string;
consumingAssetPattern?: string;
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -342,7 +353,7 @@ export const useDagRunServiceGetDagRuns = <TData =
Common.DagRunServiceGetDagRun
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
ru [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfte [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if
requested.
* 🚧 This is an experimental endpoint and may change or be removed without
notice.Successful response are streamed as newline-delimited JSON (NDJSON).
Each line is a JSON object representing the DAG run state.
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 161186260f7..8f61c1729f1 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -261,8 +261,18 @@ export const
useDagRunServiceGetUpstreamAssetEventsSuspense = <TData = Common.Da
* Get all DAG Runs.
*
* This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs for
all DAGs.
+*
+* Supports two pagination modes:
+*
+* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
+*
+* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+* on the first page.
* @param data The data for the request.
* @param data.dagId
+* @param data.cursor Cursor for keyset-based pagination. Pass an empty string
for the first page, then use ``next_cursor`` from the response. When ``cursor``
is provided, ``offset`` is ignored.
* @param data.limit
* @param data.offset
* @param data.runAfterGte
@@ -303,10 +313,11 @@ export const
useDagRunServiceGetUpstreamAssetEventsSuspense = <TData = Common.Da
* @returns DAGRunCollectionResponse Successful Response
* @throws ApiError
*/
-export const useDagRunServiceGetDagRunsSuspense = <TData =
Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
runAfte [...]
+export const useDagRunServiceGetDagRunsSuspense = <TData =
Common.DagRunServiceGetDagRunsDefaultResponse, TError = unknown, TQueryKey
extends Array<unknown> = unknown[]>({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfterGte,
[...]
bundleVersion?: string;
confContains?: string;
consumingAssetPattern?: string;
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagVersion?: number[];
@@ -342,7 +353,7 @@ export const useDagRunServiceGetDagRunsSuspense = <TData =
Common.DagRunServiceG
updatedAtGte?: string;
updatedAtLt?: string;
updatedAtLte?: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, runAfte [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseDagRunServiceGetDagRunsKeyFn({ bundleVersion, confContains,
consumingAssetPattern, cursor, dagId, dagIdPattern, dagVersion, durationGt,
durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt,
endDateLte, limit, logicalDateGt, logicalDateGte, logicalDateLt,
logicalDateLte, offset, orderBy, partitionKeyPattern, runAfterGt, [...]
/**
* Experimental: Wait for a dag run to complete, and return task results if
requested.
* 🚧 This is an experimental endpoint and may change or be removed without
notice.Successful response are streamed as newline-delimited JSON (NDJSON).
Each line is a JSON object representing the DAG run state.
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 3ae75c2db47..b97d012d8d7 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
@@ -2555,14 +2555,53 @@ export const $DAGRunCollectionResponse = {
title: 'Dag Runs'
},
total_entries: {
- type: 'integer',
- title: 'Total Entries'
+ anyOf: [
+ {
+ type: 'integer'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Total Entries',
+ description: 'Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.'
+ },
+ next_cursor: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Next Cursor',
+ description: 'Token pointing to the next page. Populated for
cursor pagination, ``null`` when using offset pagination or when there is no
next page.'
+ },
+ previous_cursor: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Previous Cursor',
+ description: 'Token pointing to the previous page. Populated for
cursor pagination, ``null`` when using offset pagination or when on the first
page.'
}
},
type: 'object',
- required: ['dag_runs', 'total_entries'],
+ required: ['dag_runs'],
title: 'DAGRunCollectionResponse',
- description: 'DAG Run Collection serializer for responses.'
+ description: `DAG Run collection response supporting both offset and
cursor pagination.
+
+A single flat model is used instead of a discriminated union
+(\`\`Annotated[Offset | Cursor, Field(discriminator=...)]\`\`) because
+the OpenAPI \`\`oneOf\`\` + \`\`discriminator\`\` construct is not handled
+correctly by \`\`@hey-api/openapi-ts\`\` /
\`\`@7nohe/openapi-react-query-codegen\`\`:
+return types degrade to \`\`unknown\`\` in JSDoc and can produce
+incorrect TypeScript types (see hey-api/openapi-ts#1613, #3270).`
} as const;
export const $DAGRunPatchBody = {
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 9d40444f7c4..ed3eaec39f8 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
@@ -975,8 +975,18 @@ export class DagRunService {
* Get all DAG Runs.
*
* This endpoint allows specifying `~` as the dag_id to retrieve Dag Runs
for all DAGs.
+ *
+ * Supports two pagination modes:
+ *
+ * **Offset (default):** use `limit` and `offset` query parameters.
Returns `total_entries`.
+ *
+ * **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
+ * When `cursor` is provided, `offset` is ignored and `total_entries` is
not returned.
+ * ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
+ * on the first page.
* @param data The data for the request.
* @param data.dagId
+ * @param data.cursor Cursor for keyset-based pagination. Pass an empty
string for the first page, then use ``next_cursor`` from the response. When
``cursor`` is provided, ``offset`` is ignored.
* @param data.limit
* @param data.offset
* @param data.runAfterGte
@@ -1025,6 +1035,7 @@ export class DagRunService {
dag_id: data.dagId
},
query: {
+ cursor: data.cursor,
limit: data.limit,
offset: data.offset,
run_after_gte: data.runAfterGte,
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 ff3e99b8cb8..79f999099ac 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
@@ -674,11 +674,29 @@ export type DAGRunClearBody = {
};
/**
- * DAG Run Collection serializer for responses.
+ * DAG Run collection response supporting both offset and cursor pagination.
+ *
+ * A single flat model is used instead of a discriminated union
+ * (``Annotated[Offset | Cursor, Field(discriminator=...)]``) because
+ * the OpenAPI ``oneOf`` + ``discriminator`` construct is not handled
+ * correctly by ``@hey-api/openapi-ts`` /
``@7nohe/openapi-react-query-codegen``:
+ * return types degrade to ``unknown`` in JSDoc and can produce
+ * incorrect TypeScript types (see hey-api/openapi-ts#1613, #3270).
*/
export type DAGRunCollectionResponse = {
dag_runs: Array<DAGRunResponse>;
- total_entries: number;
+ /**
+ * Total number of matching items. Populated for offset pagination,
``null`` when using cursor pagination.
+ */
+ total_entries?: number | null;
+ /**
+ * Token pointing to the next page. Populated for cursor pagination,
``null`` when using offset pagination or when there is no next page.
+ */
+ next_cursor?: string | null;
+ /**
+ * Token pointing to the previous page. Populated for cursor pagination,
``null`` when using offset pagination or when on the first page.
+ */
+ previous_cursor?: string | null;
};
/**
@@ -2647,6 +2665,10 @@ export type GetDagRunsData = {
* Filter by consuming asset name or URI using pattern matching
*/
consumingAssetPattern?: string | null;
+ /**
+ * Cursor for keyset-based pagination. Pass an empty string for the first
page, then use ``next_cursor`` from the response. When ``cursor`` is provided,
``offset`` is ignored.
+ */
+ cursor?: string | null;
dagId: string;
/**
* SQL LIKE expression — use `%` / `_` wildcards (e.g. `%customer_%`). or
the pipe `|` operator for OR logic (e.g. `dag1 | dag2`). Regular expressions
are **not** supported.
diff --git a/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
b/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
index 08eb817db5b..e9c979946f9 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagRuns.tsx
@@ -204,11 +204,11 @@ export const DagRuns = () => {
partition_key: false,
},
});
- const { pagination, sorting } = tableURLState;
+ const { cursor, pagination, sorting } = tableURLState;
const [sort] = sorting;
const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] :
["-run_after"];
- const { pageIndex, pageSize } = pagination;
+ const { pageSize } = pagination;
const filteredState = searchParams.get(STATE_PARAM);
const filteredType = searchParams.get(RUN_TYPE_PARAM);
const filteredRunIdPattern = searchParams.get(RUN_ID_PATTERN_PARAM);
@@ -237,6 +237,7 @@ export const DagRuns = () => {
bundleVersion: bundleVersion ?? undefined,
confContains: confContains !== null && confContains !== "" ?
confContains : undefined,
consumingAssetPattern: filteredConsumingAsset ?? undefined,
+ cursor: cursor ?? "",
dagId: dagId ?? "~",
dagIdPattern: filteredDagIdPattern ?? undefined,
dagVersion:
@@ -248,7 +249,6 @@ export const DagRuns = () => {
limit: pageSize,
logicalDateGte: logicalDateGte ?? undefined,
logicalDateLte: logicalDateLte ?? undefined,
- offset: pageIndex * pageSize,
orderBy,
partitionKeyPattern: partitionKeyPattern ?? undefined,
runAfterGte: runAfterGte ?? undefined,
@@ -270,6 +270,9 @@ export const DagRuns = () => {
const columns = runColumns(translate, dagId);
+ const nextCursor = data?.next_cursor ?? undefined;
+ const previousCursor = data?.previous_cursor ?? undefined;
+
return (
<>
<DagRunsFilters dagId={dagId} />
@@ -280,8 +283,9 @@ export const DagRuns = () => {
initialState={tableURLState}
isLoading={isLoading}
modelName="common:dagRun"
+ nextCursor={nextCursor}
onStateChange={setTableURLState}
- total={data?.total_entries}
+ previousCursor={previousCursor}
/>
</>
);
diff --git
a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
index 2958472290d..69ab5c7fa54 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
@@ -330,8 +330,8 @@ export const TaskInstances = () => {
},
);
- const nextCursor = data && "next_cursor" in data ? data.next_cursor :
undefined;
- const previousCursor = data && "previous_cursor" in data ?
data.previous_cursor : undefined;
+ const nextCursor = data?.next_cursor ?? undefined;
+ const previousCursor = data?.previous_cursor ?? undefined;
const { allRowsSelected, clearSelections, handleRowSelect, handleSelectAll,
selectedRows } =
useRowSelection({
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
b/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
index 99fe13b3c60..90d4b78b6da 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
@@ -18,6 +18,7 @@
from __future__ import annotations
import re
+from types import SimpleNamespace
from typing import Annotated
import pytest
@@ -31,6 +32,7 @@ from airflow.api_fastapi.common.parameters import (
filter_param_factory,
)
from airflow.models import DagModel, DagRun, Log
+from airflow.models.errors import ParseImportError
class TestFilterParam:
@@ -106,6 +108,39 @@ class TestSortParam:
):
param.to_orm(None)
+ def test_aliased_sort_resolves_row_value_via_to_replace(self):
+ """
+ Cursor encoding must read the sort value through ``to_replace`` when
the
+ user-facing sort name is an alias (e.g. ``dag_run_id`` → ``run_id``).
A naive
+ ``getattr(row, user_facing_name)`` would silently yield ``None`` and
break
+ keyset pagination on the next page.
+ """
+ param = SortParam(["id", "run_id"], DagRun, {"dag_run_id":
"run_id"}).set_value(["dag_run_id"])
+ attr_names = [name for name, _col, _desc in
param.get_resolved_columns()]
+ assert attr_names == ["dag_run_id", "id"]
+
+ row = SimpleNamespace(run_id="manual__2026-04-22", id=42)
+ assert param.row_value(row, "dag_run_id") == "manual__2026-04-22"
+ assert param.row_value(row, "id") == 42
+
+ def test_row_value_raises_on_column_form_to_replace(self):
+ """
+ Column-form ``to_replace`` is not supported by cursor encoding. The
helper must
+ fail loudly so a future endpoint doesn't silently ship ``None`` cursor
tokens.
+ """
+ param = SortParam(["dag_id"], DagModel, {"last_run_state":
DagRun.state}).set_value(
+ ["last_run_state"]
+ )
+ row = SimpleNamespace(id="test_dag")
+ with pytest.raises(NotImplementedError, match="column-form
``to_replace``"):
+ param.row_value(row, "last_run_state")
+
+ def test_primary_key_is_not_duplicated_when_alias_maps_to_pk(self):
+ """Sorting by an alias that resolves to the PK must not append the PK
a second time."""
+ param = SortParam(["id"], ParseImportError, {"import_error_id":
"id"}).set_value(["import_error_id"])
+ resolved = param.get_resolved_columns()
+ assert [name for name, _col, _desc in resolved] == ["import_error_id"]
+
class TestSearchParam:
def test_to_orm_single_value(self):
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index fdf480723f4..4858748ec89 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -17,6 +17,7 @@
from __future__ import annotations
+import math
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from unittest import mock
@@ -505,6 +506,135 @@ class TestGetDagRuns:
assert response.status_code == 422
assert response.json()["detail"] == expected_detail
+ @pytest.mark.parametrize(
+ "order_by",
+ [
+ "id",
+ "dag_run_id",
+ "logical_date",
+ "-run_after",
+ ], # test with multiple ordering fields (alias, non-alias, datetime,
non-datetime)
+ )
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_first_two_page(self, test_client, order_by):
+ """First page with cursor='' and second page fetched via the returned
next_cursor."""
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 2, "order_by": order_by, "cursor": ""},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ assert body["next_cursor"] is not None
+ assert body["previous_cursor"] is None
+ assert body["total_entries"] is None
+ assert len(body["dag_runs"]) == 2
+
+ response2 = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 2, "order_by": order_by, "cursor":
body["next_cursor"]},
+ )
+ assert response2.status_code == 200, response2.json()
+ body2 = response2.json()
+ assert body2["previous_cursor"] is not None
+ assert body2["total_entries"] is None
+ assert len(body2["dag_runs"]) == 2
+ first_page_ids = {(r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"]}
+ second_page_ids = {(r["dag_id"], r["dag_run_id"]) for r in
body2["dag_runs"]}
+ assert first_page_ids.isdisjoint(second_page_ids)
+
+ @pytest.mark.parametrize(
+ "order_by",
+ ["id", "dag_run_id", "logical_date", "-run_after"],
+ )
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_returns_cursor_response(self, test_client,
order_by):
+ """When cursor param is provided, response has cursor fields and no
total_entries."""
+ response1 = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 2, "order_by": order_by, "cursor": ""},
+ )
+ assert response1.status_code == 200
+ body1 = response1.json()
+ assert body1["total_entries"] is None
+ assert len(body1["dag_runs"]) == 2
+ next_cursor = body1["next_cursor"]
+ assert next_cursor is not None
+
+ # Second (last) page using next_cursor from first page — only 2 dag
runs remain
+ response2 = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 100, "cursor": next_cursor, "order_by": order_by},
+ )
+ assert response2.status_code == 200, response2.json()
+ body2 = response2.json()
+ assert body2["next_cursor"] is None
+ assert body2["previous_cursor"] is not None
+ assert body2["total_entries"] is None
+
+ @pytest.mark.parametrize(
+ "order_by",
+ ["id", "dag_run_id", "logical_date", "-run_after"],
+ )
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_forward_and_backward_consistency(self,
test_client, order_by):
+ """Walk all pages forward via next_cursor, then backward via
previous_cursor, and compare."""
+ total_runs = 4 # 4 dag runs are created by the setup fixture
+ page_size = 2
+ max_pages = math.ceil(total_runs / page_size)
+
+ forward_ids: list[tuple[str, str]] = []
+ forward_pages: list[dict] = []
+ cursor_token = ""
+ for _ in range(max_pages):
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": page_size, "order_by": order_by, "cursor":
cursor_token},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ assert body["total_entries"] is None
+ forward_pages.append(body)
+ forward_ids.extend((r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"])
+
+ cursor_token = body.get("next_cursor")
+ if cursor_token is None:
+ break
+
+ assert len(forward_ids) == total_runs
+ assert len(forward_ids) == len(set(forward_ids)), "Forward pages
should not overlap"
+ assert len(forward_pages) == max_pages
+
+ assert forward_pages[0]["previous_cursor"] is None
+ assert forward_pages[-1]["next_cursor"] is None
+
+ backward_ids: list[tuple[str, str]] = []
+ cursor_token = forward_pages[-1]["previous_cursor"]
+ assert cursor_token is not None
+
+ for _ in range(max_pages):
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": page_size, "order_by": order_by, "cursor":
cursor_token},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ backward_ids = [(r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"]] + backward_ids
+
+ cursor_token = body.get("previous_cursor")
+ if cursor_token is None:
+ break
+
+ all_backward = backward_ids + [(r["dag_id"], r["dag_run_id"]) for r in
forward_pages[-1]["dag_runs"]]
+ assert all_backward == forward_ids
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_invalid_token(self, test_client):
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"cursor": "this-is-not-valid", "order_by": "id"},
+ )
+ assert response.status_code == 400
+
@pytest.mark.parametrize(
("dag_id", "query_params", "expected_dag_id_list"),
[
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index a5a061a9cda..4750cc528f7 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -1978,11 +1978,38 @@ class DAGCollectionResponse(BaseModel):
class DAGRunCollectionResponse(BaseModel):
"""
- DAG Run Collection serializer for responses.
+ DAG Run collection response supporting both offset and cursor pagination.
+
+ A single flat model is used instead of a discriminated union
+ (``Annotated[Offset | Cursor, Field(discriminator=...)]``) because
+ the OpenAPI ``oneOf`` + ``discriminator`` construct is not handled
+ correctly by ``@hey-api/openapi-ts`` /
``@7nohe/openapi-react-query-codegen``:
+ return types degrade to ``unknown`` in JSDoc and can produce
+ incorrect TypeScript types (see hey-api/openapi-ts#1613, #3270).
"""
dag_runs: Annotated[list[DAGRunResponse], Field(title="Dag Runs")]
- total_entries: Annotated[int, Field(title="Total Entries")]
+ total_entries: Annotated[
+ int | None,
+ Field(
+ description="Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.",
+ title="Total Entries",
+ ),
+ ] = None
+ next_cursor: Annotated[
+ str | None,
+ Field(
+ description="Token pointing to the next page. Populated for cursor
pagination, ``null`` when using offset pagination or when there is no next
page.",
+ title="Next Cursor",
+ ),
+ ] = None
+ previous_cursor: Annotated[
+ str | None,
+ Field(
+ description="Token pointing to the previous page. Populated for
cursor pagination, ``null`` when using offset pagination or when on the first
page.",
+ title="Previous Cursor",
+ ),
+ ] = None
class DAGWarningCollectionResponse(BaseModel):
