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 d4407263c7f Pass try_number to extra links API (#65661)
d4407263c7f is described below
commit d4407263c7ff2a3be125bbfcb99fddab08034079
Author: Shashwati Bhattacharyaa <[email protected]>
AuthorDate: Fri May 1 00:07:21 2026 +0530
Pass try_number to extra links API (#65661)
* Forward try_number to extra links API for correct retry links
* Rename newsfragment to match PR number
* Add tryNumber to generated OpenAPI query hooks
* Fix JSDoc param annotations for tryNumber in query hooks
* Address review: read tryNumber from searchParams, add comment, remove
newsfragment
* Address review: use TaskInstanceHistory for past tries to ensure
validation and immutability
* Fix ruff: remove duplicate test method and fix line length
* Optimize: query TaskInstance before DAG parsing to fail fast on missing TI
* Fix test: update missing_dag expected response after TI-first query
reorder
---------
Co-authored-by: Shashwati <[email protected]>
---
.../core_api/openapi/v2-rest-api-generated.yaml | 8 +++
.../core_api/routes/public/extra_links.py | 41 +++++++++---
.../src/airflow/models/taskinstancehistory.py | 8 +++
.../src/airflow/ui/openapi-gen/queries/common.ts | 10 +--
.../ui/openapi-gen/queries/ensureQueryData.ts | 12 ++--
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 12 ++--
.../src/airflow/ui/openapi-gen/queries/queries.ts | 12 ++--
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 12 ++--
.../ui/openapi-gen/requests/services.gen.ts | 8 ++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 1 +
.../ui/src/pages/TaskInstance/ExtraLinks.tsx | 7 +-
.../core_api/routes/public/test_extra_links.py | 76 +++++++++++++++++++++-
12 files changed, 174 insertions(+), 33 deletions(-)
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 bcc40ff1886..d980fda4b5b 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
@@ -4640,6 +4640,14 @@ paths:
type: integer
default: -1
title: Map Index
+ - name: try_number
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: integer
+ - type: 'null'
+ title: Try Number
responses:
'200':
description: Successful Response
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py
index 83d17b404ad..b362d318728 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/extra_links.py
@@ -47,19 +47,14 @@ def get_extra_links(
session: SessionDep,
dag_bag: DagBagDep,
map_index: int = -1,
+ try_number: int | None = None,
) -> ExtraLinkCollectionResponse:
"""Get extra links for task instance."""
from airflow.models.taskinstance import TaskInstance
+ from airflow.models.taskinstancehistory import TaskInstanceHistory
dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id))
- dag = get_dag_for_run_or_latest_version(dag_bag, dag_run, dag_id, session)
-
- try:
- task = dag.get_task(task_id)
- except TaskNotFound:
- raise HTTPException(status.HTTP_404_NOT_FOUND, f"Task with ID =
{task_id} not found")
-
ti = session.scalar(
select(TaskInstance).where(
TaskInstance.dag_id == dag_id,
@@ -75,8 +70,38 @@ def get_extra_links(
"TaskInstance not found",
)
+ dag = get_dag_for_run_or_latest_version(dag_bag, dag_run, dag_id, session)
+
+ try:
+ task = dag.get_task(task_id)
+ except TaskNotFound:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, f"Task with ID =
{task_id} not found")
+
+ # Resolve which object to use for link generation. For the current try we
use
+ # the live TI; for past tries we fetch the immutable TaskInstanceHistory
record,
+ # which also validates that the requested try_number actually exists.
+ if try_number is not None and try_number != ti.try_number:
+ tih = session.scalar(
+ select(TaskInstanceHistory).where(
+ TaskInstanceHistory.dag_id == dag_id,
+ TaskInstanceHistory.task_id == task_id,
+ TaskInstanceHistory.run_id == dag_run_id,
+ TaskInstanceHistory.map_index == map_index,
+ TaskInstanceHistory.try_number == try_number,
+ )
+ )
+ if not tih:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"TaskInstanceHistory not found for try_number={try_number}",
+ )
+ ti_for_links = tih
+ else:
+ ti_for_links = ti
+
all_extra_link_pairs = (
- (link_name, task.get_extra_links(ti, link_name)) for link_name in
task.extra_links
+ (link_name, task.get_extra_links(ti_for_links, link_name))
+ for link_name in task.extra_links # type: ignore[arg-type]
)
all_extra_links = {link_name: link_url or None for link_name, link_url in
sorted(all_extra_link_pairs)}
diff --git a/airflow-core/src/airflow/models/taskinstancehistory.py
b/airflow-core/src/airflow/models/taskinstancehistory.py
index ee192624123..250b7f0d36f 100644
--- a/airflow-core/src/airflow/models/taskinstancehistory.py
+++ b/airflow-core/src/airflow/models/taskinstancehistory.py
@@ -58,6 +58,7 @@ if TYPE_CHECKING:
from airflow.models import DagRun
from airflow.models.taskinstance import TaskInstance
+ from airflow.models.taskinstancekey import TaskInstanceKey
class TaskInstanceHistory(Base):
@@ -179,6 +180,13 @@ class TaskInstanceHistory(Base):
"""Alias for primary key field to support TaskInstance."""
return self.task_instance_id
+ @property
+ def key(self) -> TaskInstanceKey:
+ """Returns a key that identifies this history record, mirroring
TaskInstance.key."""
+ from airflow.models.taskinstancekey import TaskInstanceKey
+
+ return TaskInstanceKey(self.dag_id, self.task_id, self.run_id,
self.try_number, self.map_index)
+
@staticmethod
@provide_session
def record_ti(ti: TaskInstance, session: Session = NEW_SESSION) -> None:
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 85f0d543617..e0976d31b72 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -381,21 +381,23 @@ export const UseEventLogServiceGetEventLogsKeyFn = ({
after, before, dagId, dagI
export type ExtraLinksServiceGetExtraLinksDefaultResponse =
Awaited<ReturnType<typeof ExtraLinksService.getExtraLinks>>;
export type ExtraLinksServiceGetExtraLinksQueryResult<TData =
ExtraLinksServiceGetExtraLinksDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useExtraLinksServiceGetExtraLinksKey =
"ExtraLinksServiceGetExtraLinks";
-export const UseExtraLinksServiceGetExtraLinksKeyFn = ({ dagId, dagRunId,
mapIndex, taskId }: {
+export const UseExtraLinksServiceGetExtraLinksKeyFn = ({ dagId, dagRunId,
mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: Array<unknown>) => [useExtraLinksServiceGetExtraLinksKey,
...(queryKey ?? [{ dagId, dagRunId, mapIndex, taskId }])];
+ tryNumber?: number;
+}, queryKey?: Array<unknown>) => [useExtraLinksServiceGetExtraLinksKey,
...(queryKey ?? [{ dagId, dagRunId, mapIndex, taskId, tryNumber }])];
export type TaskInstanceServiceGetExtraLinksDefaultResponse =
Awaited<ReturnType<typeof TaskInstanceService.getExtraLinks>>;
export type TaskInstanceServiceGetExtraLinksQueryResult<TData =
TaskInstanceServiceGetExtraLinksDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useTaskInstanceServiceGetExtraLinksKey =
"TaskInstanceServiceGetExtraLinks";
-export const UseTaskInstanceServiceGetExtraLinksKeyFn = ({ dagId, dagRunId,
mapIndex, taskId }: {
+export const UseTaskInstanceServiceGetExtraLinksKeyFn = ({ dagId, dagRunId,
mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: Array<unknown>) => [useTaskInstanceServiceGetExtraLinksKey,
...(queryKey ?? [{ dagId, dagRunId, mapIndex, taskId }])];
+ tryNumber?: number;
+}, queryKey?: Array<unknown>) => [useTaskInstanceServiceGetExtraLinksKey,
...(queryKey ?? [{ dagId, dagRunId, mapIndex, taskId, tryNumber }])];
export type TaskInstanceServiceGetTaskInstanceDefaultResponse =
Awaited<ReturnType<typeof TaskInstanceService.getTaskInstance>>;
export type TaskInstanceServiceGetTaskInstanceQueryResult<TData =
TaskInstanceServiceGetTaskInstanceDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const useTaskInstanceServiceGetTaskInstanceKey =
"TaskInstanceServiceGetTaskInstance";
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 ca01ec0ea6b..6668c18f2ec 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -769,15 +769,17 @@ export const ensureUseEventLogServiceGetEventLogsData =
(queryClient: QueryClien
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseExtraLinksServiceGetExtraLinksData = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId }: {
+export const ensureUseExtraLinksServiceGetExtraLinksData = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}) => queryClient.ensureQueryData({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }), queryFn: () => ExtraLinksService.getExtraLinks({ dagId, dagRunId,
mapIndex, taskId }) });
+ tryNumber?: number;
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }), queryFn: () => ExtraLinksService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId, tryNumber }) });
/**
* Get Extra Links
* Get extra links for task instance.
@@ -786,15 +788,17 @@ export const ensureUseExtraLinksServiceGetExtraLinksData
= (queryClient: QueryCl
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseTaskInstanceServiceGetExtraLinksData = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId }: {
+export const ensureUseTaskInstanceServiceGetExtraLinksData = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}) => queryClient.ensureQueryData({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }), queryFn: () => TaskInstanceService.getExtraLinks({ dagId, dagRunId,
mapIndex, taskId }) });
+ tryNumber?: number;
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }), queryFn: () => TaskInstanceService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId, tryNumber }) });
/**
* Get Task Instance
* Get task instance.
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 855bf502b3e..85bf0b04e61 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -769,15 +769,17 @@ export const prefetchUseEventLogServiceGetEventLogs =
(queryClient: QueryClient,
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseExtraLinksServiceGetExtraLinks = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId }: {
+export const prefetchUseExtraLinksServiceGetExtraLinks = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}) => queryClient.prefetchQuery({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }), queryFn: () => ExtraLinksService.getExtraLinks({ dagId, dagRunId,
mapIndex, taskId }) });
+ tryNumber?: number;
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }), queryFn: () => ExtraLinksService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId, tryNumber }) });
/**
* Get Extra Links
* Get extra links for task instance.
@@ -786,15 +788,17 @@ export const prefetchUseExtraLinksServiceGetExtraLinks =
(queryClient: QueryClie
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseTaskInstanceServiceGetExtraLinks = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId }: {
+export const prefetchUseTaskInstanceServiceGetExtraLinks = (queryClient:
QueryClient, { dagId, dagRunId, mapIndex, taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}) => queryClient.prefetchQuery({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }), queryFn: () => TaskInstanceService.getExtraLinks({ dagId, dagRunId,
mapIndex, taskId }) });
+ tryNumber?: number;
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }), queryFn: () => TaskInstanceService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId, tryNumber }) });
/**
* Get Task Instance
* Get task instance.
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 ae14c88bec4..3d3a4e14a02 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -769,15 +769,17 @@ export const useEventLogServiceGetEventLogs = <TData =
Common.EventLogServiceGet
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const useExtraLinksServiceGetExtraLinks = <TData =
Common.ExtraLinksServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId }: {
+export const useExtraLinksServiceGetExtraLinks = <TData =
Common.ExtraLinksServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }, queryKey), queryFn: () => ExtraLinksService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId }) as TData, ...options });
+ tryNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }, queryKey), queryFn: () =>
ExtraLinksService.getExtraLinks({ dagId, dagRunId, mapIndex, taskId, tryNumber
}) as TData, ...options });
/**
* Get Extra Links
* Get extra links for task instance.
@@ -786,15 +788,17 @@ export const useExtraLinksServiceGetExtraLinks = <TData =
Common.ExtraLinksServi
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const useTaskInstanceServiceGetExtraLinks = <TData =
Common.TaskInstanceServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId }: {
+export const useTaskInstanceServiceGetExtraLinks = <TData =
Common.TaskInstanceServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }, queryKey), queryFn: () => TaskInstanceService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId }) as TData, ...options });
+ tryNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }, queryKey), queryFn: () =>
TaskInstanceService.getExtraLinks({ dagId, dagRunId, mapIndex, taskId,
tryNumber }) as TData, ...options });
/**
* Get Task Instance
* Get task instance.
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 c3553e24046..d94138ba2e6 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -769,15 +769,17 @@ export const useEventLogServiceGetEventLogsSuspense =
<TData = Common.EventLogSe
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const useExtraLinksServiceGetExtraLinksSuspense = <TData =
Common.ExtraLinksServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId }: {
+export const useExtraLinksServiceGetExtraLinksSuspense = <TData =
Common.ExtraLinksServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }, queryKey), queryFn: () => ExtraLinksService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId }) as TData, ...options });
+ tryNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseExtraLinksServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }, queryKey), queryFn: () =>
ExtraLinksService.getExtraLinks({ dagId, dagRunId, mapIndex, taskId, tryNumber
}) as TData, ...options });
/**
* Get Extra Links
* Get extra links for task instance.
@@ -786,15 +788,17 @@ export const useExtraLinksServiceGetExtraLinksSuspense =
<TData = Common.ExtraLi
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+* @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
-export const useTaskInstanceServiceGetExtraLinksSuspense = <TData =
Common.TaskInstanceServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId }: {
+export const useTaskInstanceServiceGetExtraLinksSuspense = <TData =
Common.TaskInstanceServiceGetExtraLinksDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, dagRunId, mapIndex,
taskId, tryNumber }: {
dagId: string;
dagRunId: string;
mapIndex?: number;
taskId: string;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId }, queryKey), queryFn: () => TaskInstanceService.getExtraLinks({ dagId,
dagRunId, mapIndex, taskId }) as TData, ...options });
+ tryNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseTaskInstanceServiceGetExtraLinksKeyFn({ dagId, dagRunId, mapIndex,
taskId, tryNumber }, queryKey), queryFn: () =>
TaskInstanceService.getExtraLinks({ dagId, dagRunId, mapIndex, taskId,
tryNumber }) as TData, ...options });
/**
* Get Task Instance
* Get task instance.
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 9f074c52592..9c68b9e1ffe 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
@@ -1950,6 +1950,7 @@ export class ExtraLinksService {
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+ * @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
@@ -1963,7 +1964,8 @@ export class ExtraLinksService {
task_id: data.taskId
},
query: {
- map_index: data.mapIndex
+ map_index: data.mapIndex,
+ try_number: data.tryNumber
},
errors: {
401: 'Unauthorized',
@@ -1985,6 +1987,7 @@ export class TaskInstanceService {
* @param data.dagRunId
* @param data.taskId
* @param data.mapIndex
+ * @param data.tryNumber
* @returns ExtraLinkCollectionResponse Successful Response
* @throws ApiError
*/
@@ -1998,7 +2001,8 @@ export class TaskInstanceService {
task_id: data.taskId
},
query: {
- map_index: data.mapIndex
+ map_index: data.mapIndex,
+ try_number: data.tryNumber
},
errors: {
401: 'Unauthorized',
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 ba2d5f4e856..bb0fa50ce2a 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
@@ -3139,6 +3139,7 @@ export type GetExtraLinksData = {
dagRunId: string;
mapIndex?: number;
taskId: string;
+ tryNumber?: number | null;
};
export type GetExtraLinksResponse = ExtraLinkCollectionResponse;
diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstance/ExtraLinks.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstance/ExtraLinks.tsx
index da4ab0342d7..b2d72bae130 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/ExtraLinks.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/ExtraLinks.tsx
@@ -18,9 +18,10 @@
*/
import { Box, Button, Heading, HStack } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
-import { useParams } from "react-router-dom";
+import { useParams, useSearchParams } from "react-router-dom";
import { useTaskInstanceServiceGetExtraLinks } from "openapi/queries";
+import { SearchParamsKeys } from "src/constants/searchParams";
type ExtraLinksProps = {
readonly refetchInterval: number | false;
@@ -39,6 +40,9 @@ const getTarget = (url: string) => {
export const ExtraLinks = ({ refetchInterval }: ExtraLinksProps) => {
const { t: translate } = useTranslation("dag");
const { dagId = "", mapIndex = "-1", runId = "", taskId = "" } = useParams();
+ const [searchParams] = useSearchParams();
+ const tryNumberParam = searchParams.get(SearchParamsKeys.TRY_NUMBER);
+ const tryNumber = tryNumberParam === null ? undefined :
parseInt(tryNumberParam, 10);
const { data } = useTaskInstanceServiceGetExtraLinks(
{
@@ -46,6 +50,7 @@ export const ExtraLinks = ({ refetchInterval }:
ExtraLinksProps) => {
dagRunId: runId,
mapIndex: parseInt(mapIndex, 10),
taskId,
+ tryNumber,
},
undefined,
{
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py
index 931a3b49a64..6816f330cfb 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py
@@ -59,6 +59,19 @@ class AirflowPluginWithOperatorLinks(AirflowPlugin):
]
+class TryNumberLink(BaseOperatorLink):
+ name = "Try Number"
+ operators = [CustomOperator]
+
+ def get_link(self, operator, *, ti_key):
+ return f"https://example.com/logs?try_number={ti_key.try_number}";
+
+
+class TryNumberPlugin(AirflowPlugin):
+ name = "try_number_plugin"
+ operator_extra_links = [TryNumberLink()]
+
+
@pytest.mark.mock_plugin_manager(plugins=[])
class TestGetExtraLinks:
dag_id = "TEST_DAG_ID"
@@ -120,13 +133,13 @@ class TestGetExtraLinks:
pytest.param(
"/dags/INVALID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
404,
- {"detail": "The Dag with ID: `INVALID` was not found"},
+ {"detail": "TaskInstance not found"},
id="missing_dag",
),
pytest.param(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/INVALID/links",
404,
- {"detail": "Task with ID = INVALID not found"},
+ {"detail": "TaskInstance not found"},
id="missing_task",
),
],
@@ -338,3 +351,62 @@ class TestGetExtraLinks:
assert isinstance(link_value, str)
# since API returns stringified value, loading it back to compare with
original payload should be true
assert json.loads(link_value) == payload
+
+ @pytest.mark.mock_plugin_manager(plugins=[TryNumberPlugin])
+ def test_should_use_try_number_when_specified(self, test_client, session):
+ from uuid import uuid4
+
+ from sqlalchemy.sql import select
+
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.models.taskinstancehistory import TaskInstanceHistory
+
+ ti = session.scalar(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == self.dag_id,
+ TaskInstance.run_id == self.dag_run_id,
+ TaskInstance.task_id == self.task_single_link,
+ TaskInstance.map_index == -1,
+ )
+ )
+ assert ti is not None
+ original_try_number = ti.try_number
+
+ # Create a TIH record for a past try (try_number must differ from live
TI).
+ # Use a new UUID as the PK to avoid a primary-key conflict with the
live TI.
+ tih = TaskInstanceHistory(ti)
+ tih.task_instance_id = uuid4()
+ tih.try_number = original_try_number + 1
+ session.add(tih)
+ session.commit()
+
+ response = test_client.get(
+
f"/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_single_link}/links",
+ params={"try_number": original_try_number + 1},
+ )
+ assert response.status_code == 200
+ assert (
+ response.json()["extra_links"]["Try Number"]
+ == f"https://example.com/logs?try_number={original_try_number + 1}"
+ )
+
+ # Verify the live TI try_number was NOT modified
+ session.expire(ti)
+ ti = session.scalar(
+ select(TaskInstance).where(
+ TaskInstance.dag_id == self.dag_id,
+ TaskInstance.run_id == self.dag_run_id,
+ TaskInstance.task_id == self.task_single_link,
+ TaskInstance.map_index == -1,
+ )
+ )
+ assert ti is not None
+ assert ti.try_number == original_try_number
+
+ @pytest.mark.mock_plugin_manager(plugins=[TryNumberPlugin])
+ def test_should_respond_404_for_nonexistent_try_number(self, test_client):
+ response = test_client.get(
+
f"/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_single_link}/links",
+ params={"try_number": 99999},
+ )
+ assert response.status_code == 404
