You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by pi...@apache.org on 2026/04/16 21:37:22 UTC
(airflow) branch main updated: Add cursor based pagination for get_task_instances endpoint (#64845)
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 e11c6037f3a Add cursor based pagination for get_task_instances endpoint (#64845)
e11c6037f3a is described below
commit e11c6037f3a8d53abb8c03e6010469133621929c
Author: Pierre Jeambrun <pi...@gmail.com>
AuthorDate: Thu Apr 16 23:37:14 2026 +0200
Add cursor based pagination for get_task_instances endpoint (#64845)
* Add cursor-based pagination to get_task_instances endpoint
- Add cursor-based (keyset) pagination as an alternative to offset-based
pagination on the get_task_instances endpoint. Offset pagination remains
the default and is not deprecated globally.
- Response uses a discriminated union: offset responses include
total_entries, cursor responses include next_cursor and previous_cursor.
- Refactor SortParam to lazily cache column resolution instead of
mutating state in to_orm.
- Move cursor helpers (encode/decode/apply) to dedicated
common/db/cursors.py module.
- Cleanly separate cursor vs offset code paths in the endpoint handler.
* Simplify cursor token and support first page without sentinel
- Remove order_by from cursor token (now just a list of values)
- Support empty string cursor for first page (no fake sentinel needed)
- Drop order_by consistency check between cursor and query param
* Small adjustments
* Adjustments
* Narrow endpoint return types and encode cursor value types
Encode type information directly into cursor tokens as
{"type": ..., "value": ...} objects, removing the fragile
column-based type guessing during deserialization.
Narrow return types for endpoints that only return offset
pagination (patch, clear, batch, mapped) so the OpenAPI spec
and generated UI client reflect the correct types. Only
get_task_instances retains the discriminated union response.
Update UI components to use the narrowed types from the spec.
* Use msgpack for cursor tokens and nested keyset predicate
Switch cursor encoding from typed JSON to msgpack for compactness.
Replace flat OR-of-prefix-equalities with nested and/or keyset
predicate for better composite index range scans. Always use
ascending PK as the final tie-breaker for stable pagination.
* Flatten TaskInstanceCollectionRes
ponse to avoid oneOf codegen issues
Replace the discriminated union (offset | cursor response types) with
a single flat model using optional fields. OpenAPI oneOf + discriminator
is not handled correctly by hey-api/openapi-ts (#1613, #3270): return
types degrade to unknown in generated TypeScript code.
* Fix UI
* Fix CI
* Fix cursor pagination boundary detection and error handling
- Fetch limit+1 rows to accurately detect last page, returning
next_cursor=null when no more results exist
- Return previous_cursor=null on the first page (when no cursor
was provided)
- Use LimitFilter in apply_filters_to_select for the +1 limit
instead of a manual .limit() call
- Raise HTTP 400 on invalid UUID in cursor token instead of
silently passing the invalid value
- Update endpoint docs and add boundary-condition test
* Fix backward cursor based pagination
---
.../src/airflow/api_fastapi/common/cursors.py | 169 +++++++++++++++++++++
.../src/airflow/api_fastapi/common/parameters.py | 53 ++++---
.../core_api/datamodels/task_instances.py | 27 +++-
.../core_api/openapi/v2-rest-api-generated.yaml | 74 ++++++++-
.../core_api/routes/public/task_instances.py | 127 ++++++++++++----
.../src/airflow/ui/openapi-gen/queries/common.ts | 5 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 19 ++-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 19 ++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 47 +++++-
.../ui/openapi-gen/requests/services.gen.ts | 15 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 26 +++-
.../components/ActionAccordion/ActionAccordion.tsx | 2 +-
.../TaskInstance/ClearGroupTaskInstanceDialog.tsx | 2 +-
.../airflow/ui/src/pages/Dag/Overview/Overview.tsx | 8 +-
.../ui/src/pages/Task/Overview/Overview.tsx | 12 +-
.../ui/src/pages/TaskInstances/TaskInstances.tsx | 2 +-
.../tests/unit/api_fastapi/common/test_cursors.py | 144 ++++++++++++++++++
.../unit/api_fastapi/common/test_parameters.py | 7 +-
.../core_api/routes/public/test_task_instances.py | 152 +++++++++++++++++-
.../src/airflowctl/api/datamodels/generated.py | 31 +++-
22 files changed, 883 insertions(+), 96 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/cursors.py b/airflow-core/src/airflow/api_fastapi/common/cursors.py
new file mode 100644
index 00000000000..0d06aa10b2e
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/cursors.py
@@ -0,0 +1,169 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Cursor-based (keyset) pagination helpers.
+
+:meta private:
+"""
+
+from __future__ import annotations
+
+import base64
+import uuid as uuid_mod
+from typing import Any
+
+import msgspec
+from fastapi import HTTPException, status
+from sqlalchemy import and_, or_
+from sqlalchemy.sql import Select
+from sqlalchemy.sql.elements import ColumnElement
+from sqlalchemy.sql.sqltypes import Uuid
+
+from airflow.api_fastapi.common.parameters import SortParam
+
+
+def _b64url_decode_padded(token: str) -> bytes:
+ padding = 4 - (len(token) % 4)
+ if padding != 4:
+ token = token + ("=" * padding)
+ return base64.urlsafe_b64decode(token.encode("ascii"))
+
+
+def _nonstrict_bound(col: ColumnElement, value: Any, is_desc: bool) -> ColumnElement[bool]:
+ """Inclusive range edge on the leading column at each nesting level (``>=`` / ``<=``)."""
+ return col <= value if is_desc else col >= value
+
+
+def _strict_bound(col: ColumnElement, value: Any, is_desc: bool) -> ColumnElement[bool]:
+ """Strict inequality for ``or_`` branches (``<`` / ``>``)."""
+ return col < value if is_desc else col > value
+
+
+def _nested_keyset_predicate(
+ resolved: list[tuple[str, ColumnElement, bool]], values: list[Any]
+) -> ColumnElement[bool]:
+ """
+ Keyset predicate for rows strictly after the cursor in ``ORDER BY`` order.
+
+ Uses nested ``and_(non-strict, or_(strict, ...))`` so leading sort keys use
+ inclusive range bounds and inner branches use strict inequalities—friendly
+ for composite index range scans. Logically equivalent to an OR-of-prefix-
+ equalities formulation.
+ """
+ n = len(resolved)
+ _, col, is_desc = resolved[n - 1]
+ inner: ColumnElement[bool] = _strict_bound(col, values[n - 1], is_desc)
+ for i in range(n - 2, -1, -1):
+ _, col_i, is_desc_i = resolved[i]
+ inner = and_(
+ _nonstrict_bound(col_i, values[i], is_desc_i),
+ or_(_strict_bound(col_i, values[i], is_desc_i), inner),
+ )
+ return inner
+
+
+def _coerce_value(column: ColumnElement, value: Any) -> Any:
+ """Normalize decoded values for SQL bind parameters (e.g. UUID columns)."""
+ if value is None or not isinstance(value, str):
+ return value
+ ctype = getattr(column, "type", None)
+ if isinstance(ctype, Uuid):
+ try:
+ return uuid_mod.UUID(value)
+ except ValueError:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid cursor token")
+ return value
+
+
+_BACKWARD_PREFIX = "~"
+
+
+def encode_cursor(row: Any, sort_param: SortParam) -> str:
+ """
+ Encode cursor token from the boundary row of a result set.
+
+ The token is a URL-safe base64 encoding of a MessagePack list of sort-key
+ values (no padding ``=``).
+ """
+ resolved = sort_param.get_resolved_columns()
+ if not resolved:
+ raise ValueError("SortParam has no resolved columns.")
+
+ parts = [getattr(row, attr_name, None) for attr_name, _col, _desc in resolved]
+ payload = msgspec.msgpack.encode(parts)
+ return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def make_backward_cursor(token: str) -> str:
+ """Prefix a cursor token with the backward direction marker (``~``)."""
+ return f"{_BACKWARD_PREFIX}{token}"
+
+
+def parse_cursor(cursor: str) -> tuple[str, bool]:
+ """
+ Parse a raw cursor string into ``(token, is_backward)``.
+
+ Strips the ``~`` prefix if present and returns whether the cursor
+ represents a backward (previous-page) direction.
+ """
+ if cursor.startswith(_BACKWARD_PREFIX):
+ return cursor[len(_BACKWARD_PREFIX) :], True
+ return cursor, False
+
+
+def decode_cursor(token: str) -> list[Any]:
+ """Decode a cursor token to the list of sort-key values."""
+ try:
+ raw = _b64url_decode_padded(token)
+ except Exception:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid cursor token")
+
+ try:
+ data: Any = msgspec.msgpack.decode(raw)
+ except Exception:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid cursor token")
+
+ if not isinstance(data, list):
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid cursor token structure")
+
+ return data
+
+
+def apply_cursor_filter(
+ statement: Select, token: str, sort_param: SortParam, *, is_backward: bool = False
+) -> Select:
+ """
+ Apply a keyset pagination WHERE clause from a cursor token.
+
+ For forward cursors the predicate selects rows strictly *after* the cursor
+ in ORDER BY order. When *is_backward* is True the ``is_desc`` flags are
+ flipped so the predicate selects rows strictly *before* the cursor in the
+ original sort order. The caller is responsible for reversing the ORDER BY
+ and the final result list when using a backward cursor.
+ """
+ raw_values = decode_cursor(token)
+
+ resolved = sort_param.get_resolved_columns()
+ if len(raw_values) != len(resolved):
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Cursor token does not match current query shape")
+
+ parsed_values = [_coerce_value(col, val) for (_, col, _), val in zip(resolved, raw_values, strict=True)]
+
+ if is_backward:
+ resolved = [(name, col, not is_desc) for name, col, is_desc in resolved]
+
+ return statement.where(_nested_keyset_predicate(resolved, parsed_values))
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters.py b/airflow-core/src/airflow/api_fastapi/common/parameters.py
index 065dff8cda8..91250b670cf 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters.py
@@ -119,7 +119,10 @@ class OffsetFilter(BaseParam[NonNegativeInt]):
return select.offset(self.value)
@classmethod
- def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:
+ def depends(
+ cls,
+ offset: NonNegativeInt = 0,
+ ) -> OffsetFilter:
return cls().set_value(offset)
@@ -281,10 +284,16 @@ class SortParam(BaseParam[list[str]]):
self.allowed_attrs = allowed_attrs
self.model = model
self.to_replace = to_replace
+ self._cached_resolution: list[tuple[str, ColumnElement, bool]] | None = None
- def to_orm(self, select: Select) -> Select:
- if self.skip_none is False:
- raise ValueError(f"Cannot set 'skip_none' to False on a {type(self)}")
+ def set_value(self, value: list[str] | None) -> Self:
+ self._cached_resolution = None
+ return super().set_value(value)
+
+ def _resolve(self) -> list[tuple[str, ColumnElement, bool]]:
+ """Resolve sort columns as (attr_name, column, is_descending) tuples. Cached after first call."""
+ if self._cached_resolution is not None:
+ return self._cached_resolution
if self.value is None:
self.value = [self.get_primary_key_string()]
@@ -296,9 +305,10 @@ class SortParam(BaseParam[list[str]]):
f"Ordering with more than {self.MAX_SORT_PARAMS} parameters is not allowed. Provided: {order_by_values}",
)
- columns: list[ColumnElement] = []
+ resolved: list[tuple[str, ColumnElement, bool]] = []
for order_by_value in order_by_values:
lstriped_orderby = order_by_value.lstrip("-")
+ attr_name = lstriped_orderby
column: Column | None = None
if self.to_replace:
replacement = self.to_replace.get(lstriped_orderby, lstriped_orderby)
@@ -316,22 +326,31 @@ class SortParam(BaseParam[list[str]]):
if column is None:
column = getattr(self.model, lstriped_orderby)
- if order_by_value.startswith("-"):
- columns.append(column.desc())
- else:
- columns.append(column.asc())
-
- # Reset default sorting
- select = select.order_by(None)
+ resolved.append((attr_name, column, order_by_value.startswith("-")))
primary_key_column = self.get_primary_key_column()
- # Always add a final discriminator to enforce deterministic ordering.
- if order_by_values and order_by_values[0].startswith("-"):
- columns.append(primary_key_column.desc())
+ pk_name = self.get_primary_key_string()
+ if not any(name == pk_name for name, _, _ in resolved):
+ pk_desc = bool(order_by_values and order_by_values[0].startswith("-"))
+ resolved.append((pk_name, primary_key_column, pk_desc))
+
+ self._cached_resolution = resolved
+ return self._cached_resolution
+
+ def to_orm(self, select: Select, *, reversed: bool = False) -> Select:
+ if self.skip_none is False:
+ raise ValueError(f"Cannot set 'skip_none' to False on a {type(self)}")
+
+ resolved = self._resolve()
+ if reversed:
+ columns = [col.asc() if is_desc else col.desc() for _, col, is_desc in resolved]
else:
- columns.append(primary_key_column.asc())
+ columns = [col.desc() if is_desc else col.asc() for _, col, is_desc in resolved]
+ return select.order_by(None).order_by(*columns)
- return select.order_by(*columns)
+ def get_resolved_columns(self) -> list[tuple[str, ColumnElement, bool]]:
+ """Return resolved sort columns as (attr_name, column_element, is_descending) tuples."""
+ return self._resolve()
def get_primary_key_column(self) -> Column:
"""Get the primary key column of the model of SortParam object."""
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
index c8e7ab2378d..c02f1f9eee5 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
@@ -83,10 +83,33 @@ class TaskInstanceResponse(BaseModel):
class TaskInstanceCollectionResponse(BaseModel):
- """Task Instance Collection serializer for responses."""
+ """
+ Task instance 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).
+ """
task_instances: Iterable[TaskInstanceResponse]
- 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 TaskDependencyResponse(BaseModel):
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 0f98b70d39b..dbd84221ac6 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
@@ -6524,10 +6524,27 @@ paths:
description: 'Get list of task instances.
- This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve
- Task Instances for all DAGs
+ This endpoint allows specifying `~` as the dag_id, dag_run_id
- and DAG runs.'
+ to retrieve task instances for all DAGs and DAG runs.
+
+
+ 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_task_instances
security:
- OAuth2PasswordBearer: []
@@ -6545,6 +6562,20 @@ paths:
schema:
type: string
title: Dag Run 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: task_id
in: query
required: false
@@ -12596,14 +12627,45 @@ components:
type: array
title: Task Instances
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:
- task_instances
- - total_entries
title: TaskInstanceCollectionResponse
- description: Task Instance Collection serializer for responses.
+ description: 'Task instance 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).'
TaskInstanceHistoryCollectionResponse:
properties:
task_instances:
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 1af56b565f3..3d3fa538b5a 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
@@ -27,13 +27,19 @@ from sqlalchemy.orm import joinedload
from sqlalchemy.sql.selectable import Select
from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
+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_dag_for_run_or_latest_version,
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.task_instances import eager_load_TI_and_TIH_for_validation
from airflow.api_fastapi.common.parameters import (
FilterOptionEnum,
@@ -65,6 +71,7 @@ from airflow.api_fastapi.common.parameters import (
search_param_factory,
)
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.base import OrmClause
from airflow.api_fastapi.core_api.datamodels.common import BulkBody, BulkResponse
from airflow.api_fastapi.core_api.datamodels.task_instance_history import (
TaskInstanceHistoryCollectionResponse,
@@ -470,13 +477,29 @@ def get_task_instances(
],
readable_ti_filter: ReadableTIFilterDep,
session: SessionDep,
+ 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.",
+ ),
) -> TaskInstanceCollectionResponse:
"""
Get list of task instances.
- This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
- and DAG runs.
+ This endpoint allows specifying `~` as the dag_id, dag_run_id
+ to retrieve task instances for all DAGs and DAG runs.
+
+ 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
dag_run = None
query = eager_load_TI_and_TIH_for_validation(select(TI))
if dag_run_id != "~":
@@ -498,40 +521,84 @@ def get_task_instances(
if dag:
task_group_id.dag = dag
+ filters: list[OrmClause] = [
+ run_after_range,
+ logical_date_range,
+ start_date_range,
+ end_date_range,
+ update_at_range,
+ duration_range,
+ state,
+ pool,
+ pool_name_pattern,
+ queue,
+ queue_name_pattern,
+ executor,
+ task_id,
+ task_display_name_pattern,
+ task_group_id,
+ dag_id_pattern,
+ run_id_pattern,
+ version_number,
+ readable_ti_filter,
+ try_number,
+ operator,
+ operator_name_pattern,
+ map_index,
+ ]
+
+ 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 of 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]
+ )
+
+ is_backward = False
+ if cursor:
+ token, is_backward = parse_cursor(cursor)
+ if is_backward:
+ task_instance_select = order_by.to_orm(task_instance_select, reversed=True)
+ task_instance_select = apply_cursor_filter(
+ task_instance_select, token, order_by, is_backward=is_backward
+ )
+
+ fetched = list(session.scalars(task_instance_select))
+ has_more = len(fetched) > page_limit
+ task_instances = fetched[:page_limit]
+
+ if is_backward:
+ task_instances.reverse()
+ has_prev = has_more
+ has_next = True
+ else:
+ has_prev = bool(cursor)
+ has_next = has_more
+
+ return TaskInstanceCollectionResponse(
+ task_instances=task_instances,
+ next_cursor=(
+ encode_cursor(task_instances[-1], order_by) if has_next and task_instances else None
+ ),
+ previous_cursor=(
+ make_backward_cursor(encode_cursor(task_instances[0], order_by))
+ if has_prev and task_instances
+ else None
+ ),
+ )
+
task_instance_select, total_entries = paginated_select(
statement=query,
- filters=[
- run_after_range,
- logical_date_range,
- start_date_range,
- end_date_range,
- update_at_range,
- duration_range,
- state,
- pool,
- pool_name_pattern,
- queue,
- queue_name_pattern,
- executor,
- task_id,
- task_display_name_pattern,
- task_group_id,
- dag_id_pattern,
- run_id_pattern,
- version_number,
- readable_ti_filter,
- try_number,
- operator,
- operator_name_pattern,
- map_index,
- ],
+ filters=filters,
order_by=order_by,
offset=offset,
limit=limit,
session=session,
)
-
- task_instances = session.scalars(task_instance_select)
+ task_instances = list(session.scalars(task_instance_select))
return TaskInstanceCollectionResponse(
task_instances=task_instances,
total_entries=total_entries,
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 849b8ba490a..ab44ca991b0 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -478,7 +478,8 @@ export const UseTaskInstanceServiceGetMappedTaskInstanceKeyFn = ({ dagId, dagRun
export type TaskInstanceServiceGetTaskInstancesDefaultResponse = Awaited<ReturnType<typeof TaskInstanceService.getTaskInstances>>;
export type TaskInstanceServiceGetTaskInstancesQueryResult<TData = TaskInstanceServiceGetTaskInstancesDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useTaskInstanceServiceGetTaskInstancesKey = "TaskInstanceServiceGetTaskInstances";
-export const UseTaskInstanceServiceGetTaskInstancesKeyFn = ({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startDateGte, startDateLt, startDateLte, [...]
+export const UseTaskInstanceServiceGetTaskInstancesKeyFn = ({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startDateGte, startDateLt, startD [...]
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagRunId: string;
@@ -524,7 +525,7 @@ export const UseTaskInstanceServiceGetTaskInstancesKeyFn = ({ dagId, dagIdPatter
updatedAtLt?: string;
updatedAtLte?: string;
versionNumber?: number[];
-}, queryKey?: Array<unknown>) => [useTaskInstanceServiceGetTaskInstancesKey, ...(queryKey ?? [{ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startDa [...]
+}, queryKey?: Array<unknown>) => [useTaskInstanceServiceGetTaskInstancesKey, ...(queryKey ?? [{ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, [...]
export type TaskInstanceServiceGetTaskInstanceTryDetailsDefaultResponse = Awaited<ReturnType<typeof TaskInstanceService.getTaskInstanceTryDetails>>;
export type TaskInstanceServiceGetTaskInstanceTryDetailsQueryResult<TData = TaskInstanceServiceGetTaskInstanceTryDetailsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useTaskInstanceServiceGetTaskInstanceTryDetailsKey = "TaskInstanceServiceGetTaskInstanceTryDetails";
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 f841cfdca4d..2cf744fc7f2 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -902,11 +902,21 @@ export const ensureUseTaskInstanceServiceGetMappedTaskInstanceData = (queryClien
* Get Task Instances
* Get list of task instances.
*
-* This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
-* and DAG runs.
+* This endpoint allows specifying `~` as the dag_id, dag_run_id
+* to retrieve task instances for all DAGs and DAG runs.
+*
+* 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.dagRunId
+* @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.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -953,7 +963,8 @@ export const ensureUseTaskInstanceServiceGetMappedTaskInstanceData = (queryClien
* @returns TaskInstanceCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseTaskInstanceServiceGetTaskInstancesData = (queryClient: QueryClient, { dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startDateG [...]
+export const ensureUseTaskInstanceServiceGetTaskInstancesData = (queryClient: QueryClient, { cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, st [...]
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagRunId: string;
@@ -999,7 +1010,7 @@ export const ensureUseTaskInstanceServiceGetTaskInstancesData = (queryClient: Qu
updatedAtLt?: string;
updatedAtLte?: string;
versionNumber?: number[];
-}) => queryClient.ensureQueryData({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, star [...]
+}) => queryClient.ensureQueryData({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDate [...]
/**
* Get Task Instance Try Details
* Get task instance details by try number.
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 7de17099b84..dabfe8452b9 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -902,11 +902,21 @@ export const prefetchUseTaskInstanceServiceGetMappedTaskInstance = (queryClient:
* Get Task Instances
* Get list of task instances.
*
-* This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
-* and DAG runs.
+* This endpoint allows specifying `~` as the dag_id, dag_run_id
+* to retrieve task instances for all DAGs and DAG runs.
+*
+* 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.dagRunId
+* @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.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -953,7 +963,8 @@ export const prefetchUseTaskInstanceServiceGetMappedTaskInstance = (queryClient:
* @returns TaskInstanceCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseTaskInstanceServiceGetTaskInstances = (queryClient: QueryClient, { dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startDateGte [...]
+export const prefetchUseTaskInstanceServiceGetTaskInstances = (queryClient: QueryClient, { cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, star [...]
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagRunId: string;
@@ -999,7 +1010,7 @@ export const prefetchUseTaskInstanceServiceGetTaskInstances = (queryClient: Quer
updatedAtLt?: string;
updatedAtLte?: string;
versionNumber?: number[];
-}) => queryClient.prefetchQuery({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt, startD [...]
+}) => queryClient.prefetchQuery({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNamePattern, runAfterGt, runAfterGte, runAfterLt, runAfterLte, runIdPattern, startDateGt [...]
/**
* Get Task Instance Try Details
* Get task instance details by try number.
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 40b61f1e1f4..ff44edc6c8e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -902,11 +902,21 @@ export const useTaskInstanceServiceGetMappedTaskInstance = <TData = Common.TaskI
* Get Task Instances
* Get list of task instances.
*
-* This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
-* and DAG runs.
+* This endpoint allows specifying `~` as the dag_id, dag_run_id
+* to retrieve task instances for all DAGs and DAG runs.
+*
+* 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.dagRunId
+* @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.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -953,7 +963,8 @@ export const useTaskInstanceServiceGetMappedTaskInstance = <TData = Common.TaskI
* @returns TaskInstanceCollectionResponse Successful Response
* @throws ApiError
*/
-export const useTaskInstanceServiceGetTaskInstances = <TData = Common.TaskInstanceServiceGetTaskInstancesDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueNameP [...]
+export const useTaskInstanceServiceGetTaskInstances = <TData = Common.TaskInstanceServiceGetTaskInstancesDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, qu [...]
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagRunId: string;
@@ -999,7 +1010,7 @@ export const useTaskInstanceServiceGetTaskInstances = <TData = Common.TaskInstan
updatedAtLt?: string;
updatedAtLte?: string;
versionNumber?: number[];
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, queueName [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, q [...]
/**
* Get Task Instance Try Details
* Get task instance details by try number.
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 e04f3f813b5..f842542869e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -902,11 +902,21 @@ export const useTaskInstanceServiceGetMappedTaskInstanceSuspense = <TData = Comm
* Get Task Instances
* Get list of task instances.
*
-* This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
-* and DAG runs.
+* This endpoint allows specifying `~` as the dag_id, dag_run_id
+* to retrieve task instances for all DAGs and DAG runs.
+*
+* 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.dagRunId
+* @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.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -953,7 +963,8 @@ export const useTaskInstanceServiceGetMappedTaskInstanceSuspense = <TData = Comm
* @returns TaskInstanceCollectionResponse Successful Response
* @throws ApiError
*/
-export const useTaskInstanceServiceGetTaskInstancesSuspense = <TData = Common.TaskInstanceServiceGetTaskInstancesDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, qu [...]
+export const useTaskInstanceServiceGetTaskInstancesSuspense = <TData = Common.TaskInstanceServiceGetTaskInstancesDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, q [...]
+ cursor?: string;
dagId: string;
dagIdPattern?: string;
dagRunId: string;
@@ -999,7 +1010,7 @@ export const useTaskInstanceServiceGetTaskInstancesSuspense = <TData = Common.Ta
updatedAtLt?: string;
updatedAtLte?: string;
versionNumber?: number[];
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, queue, q [...]
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseTaskInstanceServiceGetTaskInstancesKeyFn({ cursor, dagId, dagIdPattern, dagRunId, durationGt, durationGte, durationLt, durationLte, endDateGt, endDateGte, endDateLt, endDateLte, executor, limit, logicalDateGt, logicalDateGte, logicalDateLt, logicalDateLte, mapIndex, offset, operator, operatorNamePattern, orderBy, pool, poolNamePattern, [...]
/**
* Get Task Instance Try Details
* Get task instance details by try number.
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 3f861d00621..e89fa2231b6 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
@@ -5327,14 +5327,53 @@ export const $TaskInstanceCollectionResponse = {
title: 'Task Instances'
},
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: ['task_instances', 'total_entries'],
+ required: ['task_instances'],
title: 'TaskInstanceCollectionResponse',
- description: 'Task Instance Collection serializer for responses.'
+ description: `Task instance 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 $TaskInstanceHistoryCollectionResponse = {
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 ce11a219fb0..bdb02550fa9 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
@@ -2319,11 +2319,21 @@ export class TaskInstanceService {
* Get Task Instances
* Get list of task instances.
*
- * This endpoint allows specifying `~` as the dag_id, dag_run_id to retrieve Task Instances for all DAGs
- * and DAG runs.
+ * This endpoint allows specifying `~` as the dag_id, dag_run_id
+ * to retrieve task instances for all DAGs and DAG runs.
+ *
+ * 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.dagRunId
+ * @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.taskId
* @param data.runAfterGte
* @param data.runAfterGt
@@ -2379,6 +2389,7 @@ export class TaskInstanceService {
dag_run_id: data.dagRunId
},
query: {
+ cursor: data.cursor,
task_id: data.taskId,
run_after_gte: data.runAfterGte,
run_after_gt: data.runAfterGt,
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 adf7d7a0cc8..70dad1e2e8a 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
@@ -1385,11 +1385,29 @@ export type TaskInletAssetReference = {
};
/**
- * Task Instance Collection serializer for responses.
+ * Task instance 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 TaskInstanceCollectionResponse = {
task_instances: Array<TaskInstanceResponse>;
- 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;
};
/**
@@ -3104,6 +3122,10 @@ export type PatchTaskInstanceByMapIndexData = {
export type PatchTaskInstanceByMapIndexResponse = TaskInstanceCollectionResponse;
export type GetTaskInstancesData = {
+ /**
+ * 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/components/ActionAccordion/ActionAccordion.tsx b/airflow-core/src/airflow/ui/src/components/ActionAccordion/ActionAccordion.tsx
index 514d93799d5..0b8920b8c5a 100644
--- a/airflow-core/src/airflow/ui/src/components/ActionAccordion/ActionAccordion.tsx
+++ b/airflow-core/src/airflow/ui/src/components/ActionAccordion/ActionAccordion.tsx
@@ -99,7 +99,7 @@ const ActionAccordion = ({ affectedTasks, groupByRunId = false, note, setNote }:
<Accordion.ItemTrigger>
<Text fontWeight="bold">
{translate("dags:runAndTaskActions.affectedTasks.title", {
- count: affectedTasks.total_entries,
+ count: affectedTasks.total_entries ?? 0,
})}
</Text>
</Accordion.ItemTrigger>
diff --git a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
index 5dc09c1f421..0b5beb490ef 100644
--- a/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx
@@ -117,7 +117,7 @@ export const ClearGroupTaskInstanceDialog = ({ onClose, open, taskInstance }: Pr
<Heading size="xl">
<strong>
{translate("dags:runAndTaskActions.clear.title", {
- type: translate("taskInstance", { count: affectedTasks.total_entries }),
+ type: translate("taskInstance", { count: affectedTasks.total_entries ?? 0 }),
})}
:
</strong>{" "}
diff --git a/airflow-core/src/airflow/ui/src/pages/Dag/Overview/Overview.tsx b/airflow-core/src/airflow/ui/src/pages/Dag/Overview/Overview.tsx
index 7ea155fe847..b284777b351 100644
--- a/airflow-core/src/airflow/ui/src/pages/Dag/Overview/Overview.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Dag/Overview/Overview.tsx
@@ -60,6 +60,8 @@ export const Overview = () => {
state: ["failed"],
});
+ const failedTaskCount = failedTasks?.total_entries ?? 0;
+
const [limit] = useLocalStorage<number>(dagRunsLimitKey(dagId ?? ""), 10);
const { data: failedRuns, isLoading: isLoadingFailedRuns } = useDagRunServiceGetDagRuns({
dagId: dagId ?? "",
@@ -94,14 +96,14 @@ export const Overview = () => {
</Box>
<HStack flexWrap="wrap">
<TrendCountButton
- colorPalette={(failedTasks?.total_entries ?? 0) === 0 ? "green" : "failed"}
- count={failedTasks?.total_entries ?? 0}
+ colorPalette={failedTaskCount === 0 ? "green" : "failed"}
+ count={failedTaskCount}
endDate={endDate}
events={(failedTasks?.task_instances ?? []).map((ti) => ({
timestamp: ti.start_date ?? ti.logical_date,
}))}
isLoading={isLoading}
- label={translate("overview.buttons.failedTask", { count: failedTasks?.total_entries ?? 0 })}
+ label={translate("overview.buttons.failedTask", { count: failedTaskCount })}
route={{
pathname: "tasks",
search: `${SearchParamsKeys.STATE}=failed`,
diff --git a/airflow-core/src/airflow/ui/src/pages/Task/Overview/Overview.tsx b/airflow-core/src/airflow/ui/src/pages/Task/Overview/Overview.tsx
index 164b34c02d7..e9fc510ff6f 100644
--- a/airflow-core/src/airflow/ui/src/pages/Task/Overview/Overview.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Task/Overview/Overview.tsx
@@ -42,7 +42,7 @@ export const Overview = () => {
const refetchInterval = useAutoRefresh({});
- const { data: failedTaskInstances, isLoading: isFailedTaskInstancesLoading } =
+ const { data: failedTaskInstancesData, isLoading: isFailedTaskInstancesLoading } =
useTaskInstanceServiceGetTaskInstances({
dagId,
dagRunId: "~",
@@ -54,6 +54,8 @@ export const Overview = () => {
taskId: Boolean(groupId) ? undefined : taskId,
});
+ const failedTaskCount = failedTaskInstancesData?.total_entries ?? 0;
+
const { data: tiData, isLoading: isLoadingTaskInstances } = useTaskInstanceServiceGetTaskInstances(
{
dagId,
@@ -84,15 +86,15 @@ export const Overview = () => {
</Box>
<HStack flexWrap="wrap">
<TrendCountButton
- colorPalette={(failedTaskInstances?.total_entries ?? 0) === 0 ? "green" : "red"}
- count={failedTaskInstances?.total_entries ?? 0}
+ colorPalette={failedTaskCount === 0 ? "green" : "red"}
+ count={failedTaskCount}
endDate={endDate}
- events={(failedTaskInstances?.task_instances ?? []).map((ti) => ({
+ events={(failedTaskInstancesData?.task_instances ?? []).map((ti) => ({
timestamp: ti.start_date ?? ti.logical_date,
}))}
isLoading={isFailedTaskInstancesLoading}
label={translate("overview.buttons.failedTaskInstance", {
- count: failedTaskInstances?.total_entries ?? 0,
+ count: failedTaskCount,
})}
route={{
pathname: "task_instances",
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 e066436afd0..a796ae23ce4 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
@@ -360,7 +360,7 @@ export const TaskInstances = () => {
isLoading={isLoading}
modelName="common:taskInstance"
onStateChange={setTableURLState}
- total={data?.total_entries}
+ total={data?.total_entries ?? undefined}
/>
<ActionBar.Root closeOnInteractOutside={false} open={Boolean(selectedRows.size)}>
<ActionBar.Content>
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_cursors.py b/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
new file mode 100644
index 00000000000..b863de9eb6f
--- /dev/null
+++ b/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
@@ -0,0 +1,144 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import base64
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import msgspec
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import select
+
+from airflow.api_fastapi.common.cursors import apply_cursor_filter, decode_cursor, encode_cursor
+from airflow.api_fastapi.common.parameters import SortParam
+from airflow.models.taskinstance import TaskInstance
+
+
+def _msgpack_cursor_token(payload: object) -> str:
+ """Match production: msgpack + base64url without padding."""
+ return base64.urlsafe_b64encode(msgspec.msgpack.encode(payload)).decode("ascii").rstrip("=")
+
+
+class TestCursorPagination:
+ """Tests for cursor-based pagination helpers."""
+
+ def _make_sort_param_with_resolved_columns(self, order_by_values=None):
+ """Build a SortParam for TaskInstance and resolve its columns."""
+ sp = SortParam(["id", "start_date", "map_index"], TaskInstance)
+ sp.set_value(order_by_values or ["map_index"])
+ sp.to_orm(select(TaskInstance))
+ return sp
+
+ def test_encode_decode_cursor_roundtrip(self):
+ sp = self._make_sort_param_with_resolved_columns(["start_date"])
+ row = MagicMock(spec=["start_date", "id"])
+ row.start_date = "2024-01-15T10:00:00+00:00"
+ row.id = "019462ab-1234-5678-9abc-def012345678"
+
+ token = encode_cursor(row, sp)
+ decoded = decode_cursor(token)
+
+ assert decoded == [
+ "2024-01-15T10:00:00+00:00",
+ "019462ab-1234-5678-9abc-def012345678",
+ ]
+
+ def test_decode_cursor_invalid_base64(self):
+ with pytest.raises(HTTPException, match="Invalid cursor token"):
+ decode_cursor("not-valid-base64!!!")
+
+ def test_decode_cursor_invalid_msgpack(self):
+ token = base64.urlsafe_b64encode(b"not-msgpack").decode().rstrip("=")
+ with pytest.raises(HTTPException, match="Invalid cursor token"):
+ decode_cursor(token)
+
+ def test_decode_cursor_not_a_list(self):
+ token = _msgpack_cursor_token({"wrong": "type"})
+ with pytest.raises(HTTPException, match="Invalid cursor token structure"):
+ decode_cursor(token)
+
+ def test_encode_cursor_works_without_prior_to_orm(self):
+ """get_resolved_columns now lazily resolves, so to_orm is no longer required before encode."""
+ sp = SortParam(["id"], TaskInstance)
+ sp.set_value(["id"])
+ row = MagicMock(spec=["id"])
+ row.id = "019462ab-1234-5678-9abc-def012345678"
+ token = encode_cursor(row, sp)
+ decoded = decode_cursor(token)
+ assert decoded == ["019462ab-1234-5678-9abc-def012345678"]
+
+ def test_apply_cursor_filter_wrong_value_count(self):
+ sp = self._make_sort_param_with_resolved_columns(["start_date"])
+ token = _msgpack_cursor_token(["only-one-value"])
+
+ with pytest.raises(HTTPException, match="does not match"):
+ apply_cursor_filter(select(TaskInstance), token, sp)
+
+ def test_apply_cursor_filter_ascending(self):
+ sp = self._make_sort_param_with_resolved_columns(["start_date"])
+ values = [
+ datetime(2024, 1, 15, 10, 0, 0, tzinfo=timezone.utc),
+ uuid.UUID("019462ab-1234-5678-9abc-def012345678"),
+ ]
+ token = _msgpack_cursor_token(values)
+
+ stmt = apply_cursor_filter(select(TaskInstance), token, sp)
+ sql = str(stmt)
+ assert ">" in sql
+
+ def test_apply_cursor_filter_descending(self):
+ sp = self._make_sort_param_with_resolved_columns(["-start_date"])
+ values = [
+ datetime(2024, 1, 15, 10, 0, 0, tzinfo=timezone.utc),
+ uuid.UUID("019462ab-1234-5678-9abc-def012345678"),
+ ]
+ token = _msgpack_cursor_token(values)
+
+ stmt = apply_cursor_filter(select(TaskInstance), token, sp)
+ sql = str(stmt)
+ assert "<" in sql
+
+ def test_sort_param_get_resolved_columns(self):
+ sp = self._make_sort_param_with_resolved_columns(["start_date"])
+ resolved = sp.get_resolved_columns()
+
+ assert len(resolved) == 2
+ assert resolved[0][0] == "start_date"
+ assert resolved[0][2] is False
+ assert resolved[1][0] == "id"
+ assert resolved[1][2] is False
+
+ def test_sort_param_get_resolved_columns_descending(self):
+ sp = self._make_sort_param_with_resolved_columns(["-start_date"])
+ resolved = sp.get_resolved_columns()
+
+ assert len(resolved) == 2
+ assert resolved[0][0] == "start_date"
+ assert resolved[0][2] is True
+ assert resolved[1][0] == "id"
+ assert resolved[1][2] is True
+
+ def test_sort_param_pk_not_duplicated_when_sorting_by_id(self):
+ sp = self._make_sort_param_with_resolved_columns(["id"])
+ resolved = sp.get_resolved_columns()
+
+ assert len(resolved) == 1
+ assert resolved[0][0] == "id"
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 5ad860477d1..40076137467 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
@@ -24,7 +24,12 @@ import pytest
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
-from airflow.api_fastapi.common.parameters import FilterParam, SortParam, _SearchParam, filter_param_factory
+from airflow.api_fastapi.common.parameters import (
+ FilterParam,
+ SortParam,
+ _SearchParam,
+ filter_param_factory,
+)
from airflow.models import DagModel, DagRun, Log
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
index 00abc961b82..ab65e665048 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import datetime as dt
import itertools
+import math
import os
from datetime import timedelta
from typing import TYPE_CHECKING, Any
@@ -1717,6 +1718,136 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
assert (num_entries_batch1 + num_entries_batch2) == ti_count
assert response_batch1 != response_batch2
+ def test_cursor_pagination_first_page(self, test_client, session):
+ """First page with cursor='' returns cursor response without needing a real token."""
+ dag_id = "example_python_operator"
+ self.create_task_instances(
+ session,
+ task_instances=[
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(5)
+ ],
+ dag_id=dag_id,
+ )
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": 3, "order_by": ["map_index"], "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["task_instances"]) == 3
+
+ def test_cursor_pagination_returns_cursor_response(self, test_client, session):
+ """When cursor param is provided, response has cursor fields and no total_entries."""
+ dag_id = "example_python_operator"
+ self.create_task_instances(
+ session,
+ task_instances=[
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(5)
+ ],
+ dag_id=dag_id,
+ )
+ # First page in cursor mode (empty cursor)
+ response1 = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": 3, "order_by": ["map_index"], "cursor": ""},
+ )
+ assert response1.status_code == 200
+ body1 = response1.json()
+ assert body1["total_entries"] is None
+ assert len(body1["task_instances"]) == 3
+ next_cursor = body1["next_cursor"]
+ assert next_cursor is not None
+
+ # Second (last) page using next_cursor from first page — only 2 TIs remain
+ response2 = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": 100, "cursor": next_cursor, "order_by": ["map_index"]},
+ )
+ assert response2.status_code == 200
+ body2 = response2.json()
+ assert body2["next_cursor"] is None
+ assert body2["previous_cursor"] is not None
+ assert body2["total_entries"] is None
+
+ def test_cursor_pagination_forward_and_backward_consistency(self, test_client, session):
+ """Walk all pages forward via next_cursor, then backward via previous_cursor, and compare."""
+ dag_id = "example_python_operator"
+ total_tis = 13
+ page_size = 4
+ max_pages = math.ceil(total_tis / page_size)
+ self.create_task_instances(
+ session,
+ task_instances=[
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(total_tis)
+ ],
+ dag_id=dag_id,
+ )
+
+ # -- Walk forward collecting pages --
+ forward_ids: list[str] = []
+ forward_pages: list[dict] = []
+ cursor_token = ""
+ for _ in range(max_pages):
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": page_size, "order_by": ["map_index"], "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(ti["id"] for ti in body["task_instances"])
+
+ cursor_token = body.get("next_cursor")
+ if cursor_token is None:
+ break
+
+ # Sanity: all TIs collected, no overlaps, multiple pages
+ assert len(forward_ids) == total_tis
+ assert len(forward_ids) == len(set(forward_ids)), "Forward pages should not overlap"
+ assert len(forward_pages) == 4
+
+ # Boundary cursors
+ assert forward_pages[0]["previous_cursor"] is None, "First page should have no previous_cursor"
+ assert forward_pages[-1]["next_cursor"] is None, "Last page should have no next_cursor"
+
+ # -- Walk backward from the last page using previous_cursor --
+ backward_ids: list[str] = []
+ cursor_token = forward_pages[-1]["previous_cursor"]
+ assert cursor_token is not None, "Last page should provide a previous_cursor"
+
+ for _ in range(max_pages):
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": page_size, "order_by": ["map_index"], "cursor": cursor_token},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ backward_ids = [ti["id"] for ti in body["task_instances"]] + backward_ids
+
+ cursor_token = body.get("previous_cursor")
+ if cursor_token is None:
+ break
+
+ # Backward walk covers all items except the last page (already collected).
+ # Order must match exactly — no re-sorting needed if pagination is correct.
+ all_backward = backward_ids + [ti["id"] for ti in forward_pages[-1]["task_instances"]]
+ assert all_backward == forward_ids, (
+ "Walking backward + last page should produce the same TIs in the same order as walking forward"
+ )
+
+ def test_cursor_pagination_invalid_token(self, test_client, session):
+ """Invalid cursor token returns 400."""
+ self.create_task_instances(session)
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"cursor": "this-is-not-valid", "order_by": ["map_index"]},
+ )
+ assert response.status_code == 400
+
def test_task_group_filter_uses_run_version_not_latest(self, test_client, dag_maker, session):
"""
Task group lookup should use the DAG version from the run, not the latest version.
@@ -4247,6 +4378,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
}
mock_set_ti_state.assert_called_once_with(
@@ -4521,6 +4654,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
},
1,
),
@@ -4657,6 +4792,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
}
_check_task_instance_note(session, response_data["task_instances"][0]["id"], ti_note_data)
@@ -4718,6 +4855,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
}
_check_task_instance_note(
@@ -4797,6 +4936,8 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
}
_check_task_instance_note(
@@ -4994,6 +5135,8 @@ class TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
}
mock_set_ti_state.assert_called_once_with(
@@ -5280,6 +5423,8 @@ class TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "next_cursor": None,
+ "previous_cursor": None,
},
1,
),
@@ -5358,7 +5503,12 @@ class TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
},
)
assert response.status_code == 200
- assert response.json() == {"task_instances": [], "total_entries": 0}
+ assert response.json() == {
+ "task_instances": [],
+ "total_entries": 0,
+ "next_cursor": None,
+ "previous_cursor": None,
+ }
class TestDeleteTaskInstance(TestTaskInstanceEndpoint):
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 6a182fdb206..e10462fa553 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -2049,11 +2049,38 @@ class TaskCollectionResponse(BaseModel):
class TaskInstanceCollectionResponse(BaseModel):
"""
- Task Instance Collection serializer for responses.
+ Task instance 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).
"""
task_instances: Annotated[list[TaskInstanceResponse], Field(title="Task Instances")]
- 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 TaskInstanceHistoryCollectionResponse(BaseModel):