This is an automated email from the ASF dual-hosted git repository.
bbovenzi 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 83b140fbcc6 Apply Gantt date filters to mapped tasks and task groups
(#75725)
83b140fbcc6 is described below
commit 83b140fbcc60030b8b161270ba284e6986302248
Author: Deepak Jain (DJ) <[email protected]>
AuthorDate: Wed Nov 11 13:53:12 2026 -0800
Apply Gantt date filters to mapped tasks and task groups (#75725)
Old tries can stretch aggregate bars across a month after clearing a run.
Date filters must also constrain those aggregates so task durations remain
readable.
Related: #63715.
Signed-off-by: Deepak Jain <[email protected]>
---
.../src/airflow/api_fastapi/common/db/dag_runs.py | 9 +
.../api_fastapi/core_api/openapi/_private_ui.yaml | 72 ++++++
.../api_fastapi/core_api/routes/ui/gantt.py | 14 +-
.../airflow/api_fastapi/core_api/routes/ui/grid.py | 81 ++++++-
.../src/airflow/ui/openapi-gen/queries/common.ts | 12 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 20 +-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 20 +-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 20 +-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 20 +-
.../ui/openapi-gen/requests/services.gen.ts | 18 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 8 +
.../ui/src/layouts/Details/Gantt/Gantt.test.tsx | 19 +-
.../airflow/ui/src/layouts/Details/Gantt/Gantt.tsx | 4 +
.../ui/src/queries/useGridTISummaries.test.tsx | 72 ++++++
.../airflow/ui/src/queries/useGridTISummaries.ts | 25 ++-
.../api_fastapi/core_api/routes/ui/test_grid.py | 245 +++++++++++++++++++++
16 files changed, 624 insertions(+), 35 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
b/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
index ed559e30225..71168ebc3e6 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
@@ -126,3 +126,12 @@ def attach_dag_versions_to_runs(dag_runs:
Sequence[DagRun], *, session: Session)
for dr in runs_needing_versions:
dr._prefetched_dag_version_ids = versions_per_run.get((dr.dag_id,
dr.run_id), {})
+
+
+def is_filtered_existing_run(dag_id: str, run_id: str, *, has_filter: bool,
session: Session) -> bool:
+ """Return whether an active filter can produce an empty result for this
run."""
+ return (
+ has_filter
+ and session.scalar(select(DagRun.run_id).where(DagRun.dag_id ==
dag_id, DagRun.run_id == run_id))
+ is not None
+ )
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index bee698ff30d..a58ea3fd517 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -2447,6 +2447,78 @@ paths:
type: string
- type: 'null'
title: Run Ids
+ - name: start_date_gte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Start Date Gte
+ - name: start_date_gt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Start Date Gt
+ - name: start_date_lte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Start Date Lte
+ - name: start_date_lt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Start Date Lt
+ - name: end_date_gte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: End Date Gte
+ - name: end_date_gt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: End Date Gt
+ - name: end_date_lte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: End Date Lte
+ - name: end_date_lt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: End Date Lt
responses:
'200':
description: "NDJSON stream \u2014 one ``GridTISummaries`` JSON
object per\
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
index 29f2e01b4f6..e7fc7c50edc 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py
@@ -25,13 +25,13 @@ from sqlalchemy import or_, select
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
from airflow.api_fastapi.common.dagbag import DagBagDep
from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.db.dag_runs import is_filtered_existing_run
from airflow.api_fastapi.common.parameters import RangeFilter,
datetime_range_filter_factory
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.datamodels.task_instances import
LoopIterationResponse
from airflow.api_fastapi.core_api.datamodels.ui.gantt import GanttResponse,
GanttTaskInstance
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.api_fastapi.core_api.security import requires_access_dag
-from airflow.models.dagrun import DagRun
from airflow.models.task_coordinates import TaskCoordinateResolver,
public_map_index_expression
from airflow.models.taskinstance import TaskInstance
from airflow.utils.state import TaskInstanceState
@@ -111,12 +111,12 @@ def get_gantt_data(
results = session.execute(query).fetchall()
if not results:
- filtered_existing_run = (
- start_date_range.is_active() or end_date_range.is_active()
- ) and session.scalar(
- select(DagRun.run_id).where(DagRun.dag_id == dag_id, DagRun.run_id
== run_id)
- ) is not None
- if not filtered_existing_run:
+ if not is_filtered_existing_run(
+ dag_id,
+ run_id,
+ has_filter=start_date_range.is_active() or
end_date_range.is_active(),
+ session=session,
+ ):
raise HTTPException(
status.HTTP_404_NOT_FOUND,
f"No task instances for dag_id={dag_id} run_id={run_id}",
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
index f6d571eea70..78e65efeda0 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
@@ -25,13 +25,13 @@ from uuid import UUID
import structlog
from fastapi import Depends, HTTPException, Query, status
from fastapi.responses import StreamingResponse
-from sqlalchemy import exists, select
+from sqlalchemy import exists, func, or_, select
from sqlalchemy.orm import Session, joinedload, load_only
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
from airflow.api_fastapi.common.dagbag import DagBagDep
from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
-from airflow.api_fastapi.common.db.dag_runs import attach_dag_versions_to_runs
+from airflow.api_fastapi.common.db.dag_runs import
attach_dag_versions_to_runs, is_filtered_existing_run
from airflow.api_fastapi.common.parameters import (
QueryDagRunRunTypesFilter,
QueryDagRunStateFilter,
@@ -84,6 +84,7 @@ from airflow.models.taskinstance import TaskInstance,
TaskInstanceNote
from airflow.serialization.definitions.taskgroup import SerializedLoopTaskGroup
from airflow.utils.helpers import chunks
from airflow.utils.session import create_session
+from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
from airflow.models.dagbag import DBDagBag
@@ -424,14 +425,18 @@ def _build_ti_summaries(
first = next(rows, None)
if first is None:
return None
- dag_version_id = first.dag_version_id or session.scalar(
- select(TaskInstance.dag_version_id)
- .where(
- TaskInstance.dag_id == dag_id,
- TaskInstance.run_id == run_id,
- TaskInstance.dag_version_id.is_not(None),
+ dag_version_id = (
+ getattr(first, "structure_version_id", None)
+ or first.dag_version_id
+ or session.scalar(
+ select(TaskInstance.dag_version_id)
+ .where(
+ TaskInstance.dag_id == dag_id,
+ TaskInstance.run_id == run_id,
+ TaskInstance.dag_version_id.is_not(None),
+ )
+ .limit(1)
)
- .limit(1)
)
serdag = _get_serdag(dag_bag, dag_id, dag_version_id, session)
if TYPE_CHECKING:
@@ -560,6 +565,10 @@ def _build_ti_summaries(
def get_grid_ti_summaries_stream(
dag_id: str,
dag_bag: DagBagDep,
+ start_date_range: Annotated[
+ RangeFilter, Depends(datetime_range_filter_factory("start_date",
TaskInstance))
+ ],
+ end_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("end_date", TaskInstance))],
run_ids: Annotated[list[str] | None, Query()] = None,
) -> StreamingResponse:
"""
@@ -573,6 +582,7 @@ def get_grid_ti_summaries_stream(
(keyed by ``dag_version_id``), which avoids repeated deserialization across
runs of the same version *and* across requests.
"""
+ has_date_filter = start_date_range.is_active() or
end_date_range.is_active()
def _generate() -> Generator[str, None, None]:
# Each iteration opens and closes its own DB session so the connection
is
@@ -589,7 +599,7 @@ def get_grid_ti_summaries_stream(
for run_id in run_ids or []:
with create_session(scoped=False) as session:
- tis = session.execute(
+ query = (
select(
TaskInstance.task_id,
TaskInstance.region_id,
@@ -601,6 +611,18 @@ def get_grid_ti_summaries_stream(
TaskInstance.retry_at,
DagVersion.version_number,
has_note_subq,
+ select(TaskInstance.dag_version_id)
+ .join(DagVersion, TaskInstance.dag_version_id ==
DagVersion.id)
+ .where(
+ TaskInstance.dag_id == dag_id,
+ TaskInstance.run_id == run_id,
+ TaskInstance.working_set.is_(True),
+ )
+ .order_by(DagVersion.version_number.desc())
+ .limit(1)
+ .correlate(None)
+ .scalar_subquery()
+ .label("structure_version_id"),
)
.outerjoin(DagVersion, TaskInstance.dag_version_id ==
DagVersion.id)
.where(TaskInstance.dag_id == dag_id)
@@ -608,6 +630,37 @@ def get_grid_ti_summaries_stream(
.order_by(TaskInstance.task_id)
.execution_options(yield_per=1000)
)
+ if has_date_filter:
+ query = query.where(
+ or_(
+ TaskInstance.state !=
TaskInstanceState.UP_FOR_RETRY, TaskInstance.state.is_(None)
+ )
+ ).execution_options(include_all_attempts=True)
+ query = start_date_range.to_orm(query)
+ query = end_date_range.to_orm(query)
+ if has_date_filter:
+ coordinate = (TaskInstance.task_id,
TaskInstance.region_id, TaskInstance.region_index)
+ attempts = (
+ query.with_only_columns(
+ *(c for c in query.selected_columns if c.key not
in {"start_date", "end_date"}),
+ func.min(TaskInstance.start_date)
+ .over(partition_by=coordinate)
+ .label("start_date"),
+
func.max(TaskInstance.end_date).over(partition_by=coordinate).label("end_date"),
+ func.row_number()
+ .over(partition_by=coordinate,
order_by=TaskInstance.try_number.desc())
+ .label("attempt_rank"),
+ )
+ .order_by(None)
+ .subquery()
+ )
+ query = (
+ select(attempts)
+ .where(attempts.c.attempt_rank == 1)
+ .order_by(attempts.c.task_id)
+ .execution_options(include_all_attempts=True,
yield_per=1000)
+ )
+ tis = session.execute(query)
summary = _build_ti_summaries(
dag_id,
run_id,
@@ -615,8 +668,12 @@ def get_grid_ti_summaries_stream(
session,
dag_bag=dag_bag,
)
- if summary is None:
- continue
+ if summary is None:
+ if not is_filtered_existing_run(
+ dag_id, run_id, has_filter=has_date_filter,
session=session
+ ):
+ continue
+ summary = {"run_id": run_id, "dag_id": dag_id,
"task_instances": []}
yield GridTISummaries.model_validate(summary).model_dump_json() +
"\n"
return StreamingResponse(content=_generate(),
media_type="application/x-ndjson")
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 3a73c01dc6b..cf8e82b897b 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -1260,10 +1260,18 @@ export const UseGridServiceGetGridRunsKeyFn = ({ dagId,
limit, offset, orderBy,
export type GridServiceGetGridTiSummariesStreamDefaultResponse =
Awaited<ReturnType<typeof GridService.getGridTiSummariesStream>>;
export type GridServiceGetGridTiSummariesStreamQueryResult<TData =
GridServiceGetGridTiSummariesStreamDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useGridServiceGetGridTiSummariesStreamKey =
"GridServiceGetGridTiSummariesStream";
-export const UseGridServiceGetGridTiSummariesStreamKeyFn = ({ dagId, runIds }:
{
+export const UseGridServiceGetGridTiSummariesStreamKeyFn = ({ dagId,
endDateGt, endDateGte, endDateLt, endDateLte, runIds, startDateGt,
startDateGte, startDateLt, startDateLte }: {
dagId: string;
+ endDateGt?: string;
+ endDateGte?: string;
+ endDateLt?: string;
+ endDateLte?: string;
runIds?: string[];
-}, queryKey?: Array<unknown>) => [useGridServiceGetGridTiSummariesStreamKey,
...(queryKey ?? [{ dagId, runIds }])];
+ startDateGt?: string;
+ startDateGte?: string;
+ startDateLt?: string;
+ startDateLte?: string;
+}, queryKey?: Array<unknown>) => [useGridServiceGetGridTiSummariesStreamKey,
...(queryKey ?? [{ dagId, endDateGt, endDateGte, endDateLt, endDateLte, runIds,
startDateGt, startDateGte, startDateLt, startDateLte }])];
export type GridServiceGetLoopSummaryDefaultResponse =
Awaited<ReturnType<typeof GridService.getLoopSummary>>;
export type GridServiceGetLoopSummaryQueryResult<TData =
GridServiceGetLoopSummaryDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useGridServiceGetLoopSummaryKey = "GridServiceGetLoopSummary";
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 d61132785ae..5c8ecaa124d 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -2504,13 +2504,29 @@ export const ensureUseGridServiceGetGridRunsData =
(queryClient: QueryClient, {
* @param data The data for the request.
* @param data.dagId
* @param data.runIds
+* @param data.startDateGte
+* @param data.startDateGt
+* @param data.startDateLte
+* @param data.startDateLt
+* @param data.endDateGte
+* @param data.endDateGt
+* @param data.endDateLte
+* @param data.endDateLt
* @returns string NDJSON stream — one ``GridTISummaries`` JSON object per
line, one per Dag run
* @throws ApiError
*/
-export const ensureUseGridServiceGetGridTiSummariesStreamData = (queryClient:
QueryClient, { dagId, runIds }: {
+export const ensureUseGridServiceGetGridTiSummariesStreamData = (queryClient:
QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, runIds,
startDateGt, startDateGte, startDateLt, startDateLte }: {
dagId: string;
+ endDateGt?: string;
+ endDateGte?: string;
+ endDateLt?: string;
+ endDateLte?: string;
runIds?: string[];
-}) => queryClient.ensureQueryData({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, runIds }), queryFn:
() => GridService.getGridTiSummariesStream({ dagId, runIds }) });
+ startDateGt?: string;
+ startDateGte?: string;
+ startDateLt?: string;
+ startDateLte?: string;
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, endDateGt,
endDateGte, endDateLt, endDateLte, runIds, startDateGt, startDateGte,
startDateLt, startDateLte }), queryFn: () =>
GridService.getGridTiSummariesStream({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, runIds, startDateGt, startDateGte, startDateLt, startDateLte }) });
/**
* Get Loop Summary
* Summarize one loop invocation from the run's live task instances.
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 beb03c8697d..ce68fd80739 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -2504,13 +2504,29 @@ export const prefetchUseGridServiceGetGridRuns =
(queryClient: QueryClient, { da
* @param data The data for the request.
* @param data.dagId
* @param data.runIds
+* @param data.startDateGte
+* @param data.startDateGt
+* @param data.startDateLte
+* @param data.startDateLt
+* @param data.endDateGte
+* @param data.endDateGt
+* @param data.endDateLte
+* @param data.endDateLt
* @returns string NDJSON stream — one ``GridTISummaries`` JSON object per
line, one per Dag run
* @throws ApiError
*/
-export const prefetchUseGridServiceGetGridTiSummariesStream = (queryClient:
QueryClient, { dagId, runIds }: {
+export const prefetchUseGridServiceGetGridTiSummariesStream = (queryClient:
QueryClient, { dagId, endDateGt, endDateGte, endDateLt, endDateLte, runIds,
startDateGt, startDateGte, startDateLt, startDateLte }: {
dagId: string;
+ endDateGt?: string;
+ endDateGte?: string;
+ endDateLt?: string;
+ endDateLte?: string;
runIds?: string[];
-}) => queryClient.prefetchQuery({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, runIds }), queryFn:
() => GridService.getGridTiSummariesStream({ dagId, runIds }) });
+ startDateGt?: string;
+ startDateGte?: string;
+ startDateLt?: string;
+ startDateLte?: string;
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, endDateGt,
endDateGte, endDateLt, endDateLte, runIds, startDateGt, startDateGte,
startDateLt, startDateLte }), queryFn: () =>
GridService.getGridTiSummariesStream({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, runIds, startDateGt, startDateGte, startDateLt, startDateLte }) });
/**
* Get Loop Summary
* Summarize one loop invocation from the run's live task instances.
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 39b7a669965..b654bded72b 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -2504,13 +2504,29 @@ export const useGridServiceGetGridRuns = <TData =
Common.GridServiceGetGridRunsD
* @param data The data for the request.
* @param data.dagId
* @param data.runIds
+* @param data.startDateGte
+* @param data.startDateGt
+* @param data.startDateLte
+* @param data.startDateLt
+* @param data.endDateGte
+* @param data.endDateGt
+* @param data.endDateLte
+* @param data.endDateLt
* @returns string NDJSON stream — one ``GridTISummaries`` JSON object per
line, one per Dag run
* @throws ApiError
*/
-export const useGridServiceGetGridTiSummariesStream = <TData =
Common.GridServiceGetGridTiSummariesStreamDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, runIds }: {
+export const useGridServiceGetGridTiSummariesStream = <TData =
Common.GridServiceGetGridTiSummariesStreamDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, endDateGt, endDateGte,
endDateLt, endDateLte, runIds, startDateGt, startDateGte, startDateLt,
startDateLte }: {
dagId: string;
+ endDateGt?: string;
+ endDateGte?: string;
+ endDateLt?: string;
+ endDateLte?: string;
runIds?: string[];
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, runIds },
queryKey), queryFn: () => GridService.getGridTiSummariesStream({ dagId, runIds
}) as TData, ...options });
+ startDateGt?: string;
+ startDateGte?: string;
+ startDateLt?: string;
+ startDateLte?: string;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, endDateGt,
endDateGte, endDateLt, endDateLte, runIds, startDateGt, startDateGte,
startDateLt, startDateLte }, queryKey), queryFn: () =>
GridService.getGridTiSummariesStream({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, runIds, startDateGt, startDateGte, startDateLt, startDateLte }) as
T [...]
/**
* Get Loop Summary
* Summarize one loop invocation from the run's live task instances.
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 40cd839b2e0..c6e5712300a 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -2504,13 +2504,29 @@ export const useGridServiceGetGridRunsSuspense = <TData
= Common.GridServiceGetG
* @param data The data for the request.
* @param data.dagId
* @param data.runIds
+* @param data.startDateGte
+* @param data.startDateGt
+* @param data.startDateLte
+* @param data.startDateLt
+* @param data.endDateGte
+* @param data.endDateGt
+* @param data.endDateLte
+* @param data.endDateLt
* @returns string NDJSON stream — one ``GridTISummaries`` JSON object per
line, one per Dag run
* @throws ApiError
*/
-export const useGridServiceGetGridTiSummariesStreamSuspense = <TData =
Common.GridServiceGetGridTiSummariesStreamDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, runIds }: {
+export const useGridServiceGetGridTiSummariesStreamSuspense = <TData =
Common.GridServiceGetGridTiSummariesStreamDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, endDateGt, endDateGte,
endDateLt, endDateLte, runIds, startDateGt, startDateGte, startDateLt,
startDateLte }: {
dagId: string;
+ endDateGt?: string;
+ endDateGte?: string;
+ endDateLt?: string;
+ endDateLte?: string;
runIds?: string[];
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, runIds },
queryKey), queryFn: () => GridService.getGridTiSummariesStream({ dagId, runIds
}) as TData, ...options });
+ startDateGt?: string;
+ startDateGte?: string;
+ startDateLt?: string;
+ startDateLte?: string;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseGridServiceGetGridTiSummariesStreamKeyFn({ dagId, endDateGt,
endDateGte, endDateLt, endDateLte, runIds, startDateGt, startDateGte,
startDateLt, startDateLte }, queryKey), queryFn: () =>
GridService.getGridTiSummariesStream({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, runIds, startDateGt, startDateGte, startDateLt, startDateLte [...]
/**
* Get Loop Summary
* Summarize one loop invocation from the run's live task instances.
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 0136dead895..20ede748e99 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
@@ -5889,6 +5889,14 @@ export class GridService {
* @param data The data for the request.
* @param data.dagId
* @param data.runIds
+ * @param data.startDateGte
+ * @param data.startDateGt
+ * @param data.startDateLte
+ * @param data.startDateLt
+ * @param data.endDateGte
+ * @param data.endDateGt
+ * @param data.endDateLte
+ * @param data.endDateLt
* @returns string NDJSON stream — one ``GridTISummaries`` JSON object per
line, one per Dag run
* @throws ApiError
*/
@@ -5900,7 +5908,15 @@ export class GridService {
dag_id: data.dagId
},
query: {
- run_ids: data.runIds
+ run_ids: data.runIds,
+ start_date_gte: data.startDateGte,
+ start_date_gt: data.startDateGt,
+ start_date_lte: data.startDateLte,
+ start_date_lt: data.startDateLt,
+ end_date_gte: data.endDateGte,
+ end_date_gt: data.endDateGt,
+ end_date_lte: data.endDateLte,
+ end_date_lt: data.endDateLt
},
errors: {
400: 'Bad Request',
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index ffed3367ccb..31e1e3402f6 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
@@ -6017,7 +6017,15 @@ export type GetGridRunsResponse =
Array<GridRunsResponse>;
export type GetGridTiSummariesStreamData = {
dagId: string;
+ endDateGt?: string | null;
+ endDateGte?: string | null;
+ endDateLt?: string | null;
+ endDateLte?: string | null;
runIds?: Array<(string)> | null;
+ startDateGt?: string | null;
+ startDateGte?: string | null;
+ startDateLt?: string | null;
+ startDateLte?: string | null;
};
export type GetGridTiSummariesStreamResponse = string;
diff --git
a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.test.tsx
b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.test.tsx
index cce6e00120f..404c080b7ca 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.test.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.test.tsx
@@ -30,7 +30,7 @@ import { BaseWrapper } from "src/utils/Wrapper";
import { Gantt } from "./Gantt";
-const { mockUseGanttServiceGetGanttData, mockUseGridRuns } = vi.hoisted(() =>
({
+const { mockUseGanttServiceGetGanttData, mockUseGridRuns,
mockUseGridTiSummariesStream } = vi.hoisted(() => ({
mockUseGanttServiceGetGanttData: vi.fn(() => ({ data: undefined, isLoading:
false })),
mockUseGridRuns: vi.fn(() => ({
data: [
@@ -44,6 +44,9 @@ const { mockUseGanttServiceGetGanttData, mockUseGridRuns } =
vi.hoisted(() => ({
],
isLoading: false,
})),
+ mockUseGridTiSummariesStream: vi.fn(() => ({
+ summariesByRunId: new Map([["run_1", { task_instances: [] }]]),
+ })),
}));
vi.mock("openapi/queries", async (importOriginal) => ({
@@ -60,9 +63,7 @@ vi.mock("src/queries/useGridStructure", () => ({
}));
vi.mock("src/queries/useGridTISummaries", () => ({
- useGridTiSummariesStream: () => ({
- summariesByRunId: new Map([["run_1", { task_instances: [] }]]),
- }),
+ useGridTiSummariesStream: mockUseGridTiSummariesStream,
}));
vi.mock("src/context/groups", () => ({
@@ -86,6 +87,7 @@ const createWrapper =
beforeEach(() => {
mockUseGanttServiceGetGanttData.mockClear();
mockUseGridRuns.mockClear();
+ mockUseGridTiSummariesStream.mockClear();
});
afterEach(cleanup);
@@ -129,6 +131,15 @@ describe("Gantt time range filter params", () => {
undefined,
expect.anything(),
);
+ expect(mockUseGridTiSummariesStream).toHaveBeenCalledWith({
+ dagId: "test_dag",
+ endDateGte: "2024-03-14T10:01:00Z",
+ endDateLte: "2024-03-14T10:10:00Z",
+ runIds: ["run_1"],
+ startDateGte: "2024-03-14T10:00:00Z",
+ startDateLte: "2024-03-14T10:05:00Z",
+ states: ["success"],
+ });
});
it("requests the full run when no date window is set", () => {
diff --git a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.tsx
b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.tsx
index 471ebe8738c..a384f815553 100644
--- a/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.tsx
+++ b/airflow-core/src/airflow/ui/src/layouts/Details/Gantt/Gantt.tsx
@@ -119,7 +119,11 @@ export const Gantt = ({
const { summariesByRunId } = useGridTiSummariesStream({
dagId,
+ endDateGte,
+ endDateLte,
runIds: runId && selectedRun ? [runId] : [],
+ startDateGte,
+ startDateLte,
states: selectedRun ? [selectedRun.state] : [],
});
const gridTiSummaries = summariesByRunId.get(runId);
diff --git
a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
index be9bf46e3fb..569e1097774 100644
--- a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
+++ b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.test.tsx
@@ -83,6 +83,78 @@ describe("useGridTiSummariesStream", () => {
(useAutoRefresh as Mock).mockReturnValue(false);
});
+ it("refetches aggregate summaries when the Gantt date window changes", async
() => {
+ const queryClient = new QueryClient();
+ const wrapper = createWrapper(queryClient);
+ const summary = {
+ dag_id: "dag_1",
+ run_id: "run_1",
+ task_instances: [
+ {
+ child_states: { success: 1 },
+ max_end_date: "2024-12-31T00:01:00Z",
+ min_start_date: "2024-12-31T00:00:00Z",
+ state: "success",
+ task_display_name: "group",
+ task_id: "group",
+ },
+ ],
+ };
+
+ mockFetch
+
.mockResolvedValueOnce(createMockResponse([`${JSON.stringify(summary)}\n`]))
+ .mockResolvedValueOnce(createMockResponse([`${JSON.stringify({
...summary, task_instances: [] })}\n`]))
+
.mockResolvedValueOnce(createMockResponse([`${JSON.stringify(summary)}\n`]));
+
+ const { rerender, result } = renderHook<
+ ReturnType<typeof useGridTiSummariesStream>,
+ { endDateLte?: string; startDateGte?: string }
+ >(
+ ({ endDateLte, startDateGte }) =>
+ useGridTiSummariesStream({
+ dagId: "dag_1",
+ endDateLte,
+ runIds: ["run_1"],
+ startDateGte,
+ }),
+ {
+ initialProps: { endDateLte: "2024-12-31T23:59:00Z", startDateGte:
"2024-12-31T00:00:00Z" },
+ wrapper,
+ },
+ );
+
+ await act(async () => {
+ await vi.runAllTimersAsync();
+ });
+
+ const firstUrl = new URL(mockFetch.mock.calls[0]?.[0] as string,
"http://localhost";);
+
+
expect(firstUrl.searchParams.get("start_date_gte")).toBe("2024-12-31T00:00:00Z");
+
expect(firstUrl.searchParams.get("end_date_lte")).toBe("2024-12-31T23:59:00Z");
+ expect(result.current.summariesByRunId.get("run_1")).toEqual(summary);
+
+ rerender({ startDateGte: "2025-01-01T00:00:00Z" });
+ await act(async () => {
+ await vi.runAllTimersAsync();
+ });
+ expect(mockFetch).toHaveBeenCalledTimes(2);
+ const secondUrl = new URL(mockFetch.mock.calls[1]?.[0] as string,
"http://localhost";);
+
+
expect(secondUrl.searchParams.get("start_date_gte")).toBe("2025-01-01T00:00:00Z");
+
expect(result.current.summariesByRunId.get("run_1")?.task_instances).toEqual([]);
+
+ rerender({});
+ await act(async () => {
+ await vi.runAllTimersAsync();
+ });
+
+ const thirdUrl = new URL(mockFetch.mock.calls[2]?.[0] as string,
"http://localhost";);
+
+ expect(thirdUrl.searchParams.has("start_date_gte")).toBe(false);
+ expect(thirdUrl.searchParams.has("end_date_lte")).toBe(false);
+ expect(result.current.summariesByRunId.get("run_1")).toEqual(summary);
+ });
+
it("streams summaries correctly on mount", async () => {
const queryClient = new QueryClient({
defaultOptions: {
diff --git a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
index 3fa06f7ec1e..fce4d4105e3 100644
--- a/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useGridTISummaries.ts
@@ -48,11 +48,19 @@ const GRID_MUTATION_WATCHED_KEYS = new Set([
*/
export const useGridTiSummariesStream = ({
dagId,
+ endDateGte,
+ endDateLte,
runIds,
+ startDateGte,
+ startDateLte,
states,
}: {
dagId: string;
+ endDateGte?: string | undefined;
+ endDateLte?: string | undefined;
runIds: Array<string>;
+ startDateGte?: string | undefined;
+ startDateLte?: string | undefined;
states?: Array<TaskInstanceState | null | undefined>;
}) => {
const queryClient = useQueryClient();
@@ -64,6 +72,19 @@ export const useGridTiSummariesStream = ({
// Stable key so the effect only re-fires when the run list actually changes.
const runIdsKey = runIds.join(",");
+ const dateFilters = new URLSearchParams();
+
+ Object.entries({
+ end_date_gte: endDateGte,
+ end_date_lte: endDateLte,
+ start_date_gte: startDateGte,
+ start_date_lte: startDateLte,
+ }).forEach(([key, value]) => {
+ if (value !== undefined) {
+ dateFilters.set(key, value);
+ }
+ });
+ const dateFiltersKey = dateFilters.toString();
// Stream (or re-stream) whenever the run list or refresh tick changes.
useEffect(() => {
@@ -79,6 +100,8 @@ export const useGridTiSummariesStream = ({
// place as fresh lines arrive rather than flashing blank.
try {
const params = new URLSearchParams(runIds.map((id) => ["run_ids",
id]));
+
+ new URLSearchParams(dateFiltersKey).forEach((value, key) =>
params.set(key, value));
const response = await
fetch(`${OpenAPI.BASE}/ui/grid/ti_summaries/${dagId}?${params}`, {
signal: abortController.signal,
});
@@ -138,7 +161,7 @@ export const useGridTiSummariesStream = ({
}
};
// eslint-disable-next-line react-hooks/exhaustive-deps -- runIdsKey
(stable join) intentionally replaces runIds array to avoid spurious re-streams
- }, [dagId, runIdsKey, refreshTick]);
+ }, [dagId, dateFiltersKey, runIdsKey, refreshTick]);
// Trigger a re-stream periodically while active runs are in flight.
useEffect(() => {
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
index cfb8e122cec..3add339c9ff 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py
@@ -1295,11 +1295,216 @@ class TestGetGridDataEndpoint:
assert by_task_id["mapped_task_group.subtask"]["retry_at"] is None
assert by_task_id["mapped_task_group"]["retry_at"] is None
+ @pytest.mark.parametrize(
+ ("filters", "expected_start"),
+ [
+ ({"start_date_gte": "2024-12-30T00:00:00Z"},
"2024-12-31T00:00:00Z"),
+ ({"end_date_lte": "2024-12-02T00:00:00Z"}, "2024-12-01T00:00:00Z"),
+ ({"start_date_gte": "2025-01-01T00:00:00Z"}, None),
+ ],
+ )
+ def test_grid_ti_summaries_filters_mapped_tasks_and_groups(
+ self, session, test_client, filters, expected_start
+ ):
+ tis = session.scalars(
+ select(TaskInstance).where(TaskInstance.dag_id == DAG_ID,
TaskInstance.run_id == "run_1")
+ ).all()
+ for ti in tis:
+ ti.start_date = timezone.datetime(2024, 12, 1)
+ ti.end_date = ti.start_date + timedelta(minutes=1)
+ if ti.task_id == f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}" and
ti.map_index == 1:
+ ti.start_date = timezone.datetime(2024, 12, 31)
+ ti.end_date = ti.start_date + timedelta(minutes=1)
+ session.commit()
+
+ response = test_client.get(f"/grid/ti_summaries/{DAG_ID}",
params={"run_ids": ["run_1"], **filters})
+ assert response.status_code == 200
+ [summary] = self._parse_ndjson(response)
+ if expected_start is None:
+ assert summary["task_instances"] == []
+ else:
+ by_task_id = {ti["task_id"]: ti for ti in
summary["task_instances"]}
+ for task_id in (TASK_GROUP_ID,
f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}"):
+ assert by_task_id[task_id]["min_start_date"] == expected_start
+ assert by_task_id[task_id]["max_end_date"] == (
+ timezone.parse(expected_start) + timedelta(minutes=1)
+ ).isoformat().replace("+00:00", "Z")
+
@staticmethod
def _parse_ndjson(response) -> list[dict]:
"""Parse NDJSON streaming response into a list of dicts."""
return [json.loads(line) for line in response.text.splitlines() if
line.strip()]
+ @pytest.mark.parametrize(
+ ("archived_state", "retry_state", "filters", "expected_state",
"expected_end"),
+ [
+ (
+ "success",
+ "success",
+ {"end_date_lte": "2024-12-02T00:00:00Z"},
+ "success",
+ "2024-12-01T00:01:00Z",
+ ),
+ (
+ "success",
+ "up_for_retry",
+ {"end_date_lte": "2024-12-02T00:00:00Z"},
+ "success",
+ "2024-12-01T00:01:00Z",
+ ),
+ (
+ "failed",
+ "success",
+ {"start_date_gte": "2024-12-01T00:00:00Z"},
+ "success",
+ "2024-12-31T00:01:00Z",
+ ),
+ ("failed", "success", {"end_date_lte": "2024-12-02T00:00:00Z"},
"failed", "2024-12-01T00:01:00Z"),
+ ],
+ )
+ def test_grid_ti_summaries_filters_archived_group_attempts(
+ self,
+ dag_maker,
+ session,
+ test_client,
+ archived_state,
+ retry_state,
+ filters,
+ expected_state,
+ expected_end,
+ ):
+ with dag_maker("filtered_group_history", serialized=True):
+ with TaskGroup(group_id="group"):
+ EmptyOperator(task_id="work")
+ run = dag_maker.create_dagrun()
+ first = run.task_instances[0]
+ first.try_number = 1
+ first.state = archived_state
+ first.start_date = timezone.datetime(2024, 12, 1)
+ first.end_date = first.start_date + timedelta(minutes=1)
+ first.archive(reason="cleared", session=session)
+ retry = TaskInstance(
+ task=dag_maker.serialized_dag.get_task("group.work"),
+ run_id=run.run_id,
+ dag_version_id=first.dag_version_id,
+ region_id=first.region_id,
+ region_index=first.region_index,
+ )
+ retry.try_number = 2
+ retry.state = retry_state
+ retry.start_date = (
+ first.start_date
+ if retry_state == TaskInstanceState.UP_FOR_RETRY
+ else timezone.datetime(2024, 12, 31)
+ )
+ retry.end_date = retry.start_date + timedelta(minutes=1)
+ session.add(retry)
+ session.commit()
+
+ response = test_client.get(
+ f"/grid/ti_summaries/{run.dag_id}",
+ params={"run_ids": [run.run_id], **filters},
+ )
+ assert response.status_code == 200
+ [summary] = self._parse_ndjson(response)
+ group = next(ti for ti in summary["task_instances"] if ti["task_id"]
== "group")
+ assert group["min_start_date"] == "2024-12-01T00:00:00Z"
+ assert group["max_end_date"] == expected_end
+ assert group["state"] == expected_state
+ assert group["child_states"] == {expected_state: 1}
+
+ @pytest.mark.parametrize("current_state", [TaskInstanceState.SUCCESS,
TaskInstanceState.RUNNING])
+ def test_filtered_mapped_retries_count_each_coordinate_once(self, session,
test_client, current_state):
+ task_id = f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}"
+ tis = session.scalars(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == DAG_ID,
+ TaskInstance.run_id == "run_1",
+ TaskInstance.task_id == task_id,
+ )
+ ).all()
+ assert {ti.region_index for ti in tis} == {0, 1, 2, 3}
+ for ti in tis:
+ ti.state = TaskInstanceState.FAILED
+ ti.try_number = 1
+ ti.start_date = timezone.datetime(2024, 12, 1)
+ ti.end_date = ti.start_date + timedelta(minutes=1)
+ current = ti.prepare_db_for_next_try(session=session)
+ current.state = current_state
+ current.start_date = timezone.datetime(2024, 12, 31)
+ current.end_date = (
+ current.start_date + timedelta(minutes=1)
+ if current_state == TaskInstanceState.SUCCESS
+ else None
+ )
+ session.commit()
+
+ response = test_client.get(
+ f"/grid/ti_summaries/{DAG_ID}",
+ params={"run_ids": ["run_1"], "start_date_gte":
"2024-12-01T00:00:00Z"},
+ )
+ assert response.status_code == 200
+ [summary] = self._parse_ndjson(response)
+ mapped = next(ti for ti in summary["task_instances"] if ti["task_id"]
== task_id)
+ assert mapped["child_states"] == {current_state: 4}
+ assert mapped["state"] == current_state
+ assert mapped["min_start_date"] == "2024-12-01T00:00:00Z"
+ expected_day = 31 if current_state == TaskInstanceState.SUCCESS else 1
+ assert mapped["max_end_date"] == f"2024-12-{expected_day:02}T00:01:00Z"
+
+ def test_filtered_archived_attempt_does_not_choose_an_old_group_structure(
+ self, dag_maker, session, test_client
+ ):
+ dag_id = "filtered_versioned_group"
+ with dag_maker(dag_id, serialized=True):
+ EmptyOperator(task_id="a")
+ run = dag_maker.create_dagrun(run_id="version_1")
+ archived = run.task_instances[0]
+ archived.try_number = 1
+ archived.state = TaskInstanceState.SUCCESS
+ archived.start_date = timezone.datetime(2024, 12, 1)
+ archived.end_date = archived.start_date + timedelta(minutes=1)
+ archived.archive(reason="cleared", session=session)
+
+ with dag_maker(dag_id, serialized=True):
+ EmptyOperator(task_id="a")
+ with TaskGroup(group_id="group"):
+ EmptyOperator(task_id="work")
+ dag_maker.create_dagrun(run_id="version_2",
logical_date=timezone.datetime(2024, 12, 31))
+ version_id = session.scalar(
+ select(DagVersion.id)
+ .where(DagVersion.dag_id == dag_id)
+ .order_by(DagVersion.version_number.desc())
+ .limit(1)
+ )
+ for task_id, day in [("a", 31), ("group.work", 1)]:
+ ti = TaskInstance(
+ task=dag_maker.serialized_dag.get_task(task_id),
+ run_id=run.run_id,
+ dag_version_id=archived.dag_version_id if task_id == "a" else
version_id,
+ region_id=archived.region_id,
+ region_index=archived.region_index,
+ )
+ ti.state = TaskInstanceState.SUCCESS
+ ti.start_date = timezone.datetime(2024, 12, day)
+ ti.end_date = ti.start_date + timedelta(minutes=1)
+ session.add(ti)
+ session.commit()
+
+ for filters in ({}, {"end_date_lte": "2024-12-02T00:00:00Z"}):
+ response = test_client.get(
+ f"/grid/ti_summaries/{dag_id}",
+ params={"run_ids": [run.run_id], **filters},
+ )
+ assert response.status_code == 200
+ [summary] = self._parse_ndjson(response)
+ group = next(ti for ti in summary["task_instances"] if
ti["task_id"] == "group")
+ assert group["min_start_date"] == "2024-12-01T00:00:00Z"
+ assert group["max_end_date"] == "2024-12-01T00:01:00Z"
+ child = next(ti for ti in summary["task_instances"] if
ti["task_id"] == "group.work")
+ assert child["state"] == TaskInstanceState.SUCCESS
+ assert group["child_states"] == {"success": 1}
+
def test_grid_ti_summaries_stream_returns_all_runs(self, session,
test_client):
"""Streaming endpoint returns one NDJSON line per requested run_id."""
session.commit()
@@ -1533,6 +1738,46 @@ class TestGetLoopSummaryEndpoint:
"loop_exit_criteria_doc": None,
}
+ def test_filtered_loop_retries_preserve_iterations(self,
executed_fixed_loop, session, test_client):
+ run = executed_fixed_loop
+ tis = session.scalars(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == run.dag_id,
+ TaskInstance.run_id == run.run_id,
+ TaskInstance.task_id == "body.work",
+ )
+ ).all()
+ assert {ti.region_index for ti in tis} == {1, 2, 3}
+ for ti in tis:
+ ti.state = TaskInstanceState.FAILED
+ ti.start_date = timezone.datetime(2024, 12, 1)
+ ti.end_date = ti.start_date + timedelta(minutes=1)
+ current = ti.prepare_db_for_next_try(session=session)
+ current.state = TaskInstanceState.SUCCESS
+ current.start_date = timezone.datetime(2024, 12, 31)
+ current.end_date = current.start_date + timedelta(minutes=1)
+ session.commit()
+
+ response = test_client.get(
+ f"/grid/ti_summaries/{run.dag_id}",
+ params={
+ "run_ids": [run.run_id],
+ "start_date_gte": "2024-12-01T00:00:00Z",
+ "end_date_lte": "2025-01-01T00:00:00Z",
+ },
+ )
+ assert response.status_code == 200
+ [summary] = TestGetGridDataEndpoint._parse_ndjson(response)
+ work = next(ti for ti in summary["task_instances"] if ti["task_id"] ==
"body.work")
+ assert work["state"] == TaskInstanceState.SUCCESS
+ body = next(ti for ti in summary["task_instances"] if ti["task_id"] ==
"body")
+ # Each iteration contains both the work task and the loop gate.
+ assert body["child_states"] == {"success": 6}
+ assert body["loop_iterations_count"] == 3
+ assert work["latest_region_index"] == 3
+ assert work["min_start_date"] == "2024-12-01T00:00:00Z"
+ assert work["max_end_date"] == "2024-12-31T00:01:00Z"
+
def test_ti_summaries_stream_rows_after_the_first(self,
executed_fixed_loop, session):
run = executed_fixed_loop
rows = session.execute(
