Add dag version lookup on get task endpoint

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -9880,7 +9880,16 @@ paths:
       tags:
       - Task
       summary: Get Task
-      description: Get simplified representation of a task.
+      description: 'Get simplified representation of a task.
+
+
+        Pass ``version_number`` (e.g. a task instance''s ``dag_version.version_number``)
+        to get the task
+
+        as defined in that Dag version. Without it the latest version is used. Returns
+        404 if that
+
+        version of the Dag does not exist.'
       operationId: get_task
       security:
       - OAuth2PasswordBearer: []
@@ -9897,6 +9906,14 @@ paths:
         required: true
         schema:
           title: Task Id
+      - name: version_number
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Version Number
       responses:
         '200':
           description: Successful Response

airflow-core/src/airflow/api_fastapi/core_api/routes/public/tasks.py

@@ -29,6 +29,7 @@
 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.exceptions import TaskNotFound
+from airflow.models.dag_version import DagVersion
 
 tasks_router = AirflowRouter(tags=["Task"], prefix="/dags/{dag_id}/tasks")
 
@@ -102,9 +103,31 @@ def get_tasks(
     ),
     dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK))],
 )
-def get_task(dag_id: str, task_id, session: SessionDep, dag_bag: DagBagDep) -> TaskResponse:
-    """Get simplified representation of a task."""
-    dag = get_latest_version_of_dag(dag_bag, dag_id, session)
+def get_task(
+    dag_id: str,
+    task_id,
+    session: SessionDep,
+    dag_bag: DagBagDep,
+    version_number: int | None = None,
+) -> TaskResponse:
+    """
+    Get simplified representation of a task.
+
+    Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+    as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+    version of the Dag does not exist.
+    """
+    if version_number is None:
+        dag = get_latest_version_of_dag(dag_bag, dag_id, session)
+    else:
+        dag_version = DagVersion.get_version(dag_id, version_number, session=session)
+        versioned_dag = None if dag_version is None else dag_bag.get_dag(dag_version.id, session=session)
+        if versioned_dag is None:
+            raise HTTPException(
+                status.HTTP_404_NOT_FOUND,
+                f"The Dag {dag_id}, version_number {version_number} was not found",
+            )
+        dag = versioned_dag
     try:
         task = dag.get_task(task_id=task_id)
     except TaskNotFound:

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -958,10 +958,11 @@ export const UseTaskServiceGetTasksKeyFn = ({ dagId, orderBy }: {
 export type TaskServiceGetTaskDefaultResponse = Awaited<ReturnType<typeof TaskService.getTask>>;
 export type TaskServiceGetTaskQueryResult<TData = TaskServiceGetTaskDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useTaskServiceGetTaskKey = "TaskServiceGetTask";
-export const UseTaskServiceGetTaskKeyFn = ({ dagId, taskId }: {
+export const UseTaskServiceGetTaskKeyFn = ({ dagId, taskId, versionNumber }: {
   dagId: string;
   taskId: unknown;
-}, queryKey?: Array<unknown>) => [useTaskServiceGetTaskKey, ...(queryKey ?? [{ dagId, taskId }])];
+  versionNumber?: number;
+}, queryKey?: Array<unknown>) => [useTaskServiceGetTaskKey, ...(queryKey ?? [{ dagId, taskId, versionNumber }])];
 export type VariableServiceGetVariableDefaultResponse = Awaited<ReturnType<typeof VariableService.getVariable>>;
 export type VariableServiceGetVariableQueryResult<TData = VariableServiceGetVariableDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useVariableServiceGetVariableKey = "VariableServiceGetVariable";

airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts

@@ -1893,16 +1893,22 @@ export const ensureUseTaskServiceGetTasksData = (queryClient: QueryClient, { dag
 /**
 * Get Task
 * Get simplified representation of a task.
+*
+* Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+* as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+* version of the Dag does not exist.
 * @param data The data for the request.
 * @param data.dagId
 * @param data.taskId
+* @param data.versionNumber
 * @returns TaskResponse Successful Response
 * @throws ApiError
 */
-export const ensureUseTaskServiceGetTaskData = (queryClient: QueryClient, { dagId, taskId }: {
+export const ensureUseTaskServiceGetTaskData = (queryClient: QueryClient, { dagId, taskId, versionNumber }: {
   dagId: string;
   taskId: unknown;
-}) => queryClient.ensureQueryData({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId }), queryFn: () => TaskService.getTask({ dagId, taskId }) });
+  versionNumber?: number;
+}) => queryClient.ensureQueryData({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId, versionNumber }), queryFn: () => TaskService.getTask({ dagId, taskId, versionNumber }) });
 /**
 * Get Variable
 * Get a variable entry.

airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts

@@ -1893,16 +1893,22 @@ export const prefetchUseTaskServiceGetTasks = (queryClient: QueryClient, { dagId
 /**
 * Get Task
 * Get simplified representation of a task.
+*
+* Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+* as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+* version of the Dag does not exist.
 * @param data The data for the request.
 * @param data.dagId
 * @param data.taskId
+* @param data.versionNumber
 * @returns TaskResponse Successful Response
 * @throws ApiError
 */
-export const prefetchUseTaskServiceGetTask = (queryClient: QueryClient, { dagId, taskId }: {
+export const prefetchUseTaskServiceGetTask = (queryClient: QueryClient, { dagId, taskId, versionNumber }: {
   dagId: string;
   taskId: unknown;
-}) => queryClient.prefetchQuery({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId }), queryFn: () => TaskService.getTask({ dagId, taskId }) });
+  versionNumber?: number;
+}) => queryClient.prefetchQuery({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId, versionNumber }), queryFn: () => TaskService.getTask({ dagId, taskId, versionNumber }) });
 /**
 * Get Variable
 * Get a variable entry.

airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts

@@ -1893,16 +1893,22 @@ export const useTaskServiceGetTasks = <TData = Common.TaskServiceGetTasksDefault
 /**
 * Get Task
 * Get simplified representation of a task.
+*
+* Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+* as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+* version of the Dag does not exist.
 * @param data The data for the request.
 * @param data.dagId
 * @param data.taskId
+* @param data.versionNumber
 * @returns TaskResponse Successful Response
 * @throws ApiError
 */
-export const useTaskServiceGetTask = <TData = Common.TaskServiceGetTaskDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, taskId }: {
+export const useTaskServiceGetTask = <TData = Common.TaskServiceGetTaskDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, taskId, versionNumber }: {
   dagId: string;
   taskId: unknown;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId }, queryKey), queryFn: () => TaskService.getTask({ dagId, taskId }) as TData, ...options });
+  versionNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId, versionNumber }, queryKey), queryFn: () => TaskService.getTask({ dagId, taskId, versionNumber }) as TData, ...options });
 /**
 * Get Variable
 * Get a variable entry.

airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts

@@ -1893,16 +1893,22 @@ export const useTaskServiceGetTasksSuspense = <TData = Common.TaskServiceGetTask
 /**
 * Get Task
 * Get simplified representation of a task.
+*
+* Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+* as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+* version of the Dag does not exist.
 * @param data The data for the request.
 * @param data.dagId
 * @param data.taskId
+* @param data.versionNumber
 * @returns TaskResponse Successful Response
 * @throws ApiError
 */
-export const useTaskServiceGetTaskSuspense = <TData = Common.TaskServiceGetTaskDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, taskId }: {
+export const useTaskServiceGetTaskSuspense = <TData = Common.TaskServiceGetTaskDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagId, taskId, versionNumber }: {
   dagId: string;
   taskId: unknown;
-}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId }, queryKey), queryFn: () => TaskService.getTask({ dagId, taskId }) as TData, ...options });
+  versionNumber?: number;
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseTaskServiceGetTaskKeyFn({ dagId, taskId, versionNumber }, queryKey), queryFn: () => TaskService.getTask({ dagId, taskId, versionNumber }) as TData, ...options });
 /**
 * Get Variable
 * Get a variable entry.

airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts

@@ -4634,9 +4634,14 @@ export class TaskService {
     /**
      * Get Task
      * Get simplified representation of a task.
+     *
+     * Pass ``version_number`` (e.g. a task instance's ``dag_version.version_number``) to get the task
+     * as defined in that Dag version. Without it the latest version is used. Returns 404 if that
+     * version of the Dag does not exist.
      * @param data The data for the request.
      * @param data.dagId
      * @param data.taskId
+     * @param data.versionNumber
      * @returns TaskResponse Successful Response
      * @throws ApiError
      */
@@ -4648,6 +4653,9 @@ export class TaskService {
                 dag_id: data.dagId,
                 task_id: data.taskId
             },
+            query: {
+                version_number: data.versionNumber
+            },
             errors: {
                 400: 'Bad Request',
                 401: 'Unauthorized',

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -5030,6 +5030,7 @@ export type GetTasksResponse = TaskCollectionResponse;
 export type GetTaskData = {
     dagId: string;
     taskId: unknown;
+    versionNumber?: number | null;
 };
 
 export type GetTaskResponse = TaskResponse;

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_tasks.py

@@ -21,6 +21,7 @@
 import pytest
 
 from airflow.api_fastapi.common.dagbag import dag_bag_from_app
+from airflow.models.dag_version import DagVersion
 from airflow.models.dagbag import DBDagBag
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.sdk import DAG
@@ -278,6 +279,40 @@ def test_should_respond_200_serialized(self, test_client, testing_dag_bundle):
         assert response.status_code == 200
         assert response.json() == expected
 
+    def test_version_number_describes_the_task_as_it_was(self, test_client, dag_maker, session):
+        """A later edit must not change how an earlier version's task is reported."""
+        dag_id = "test_versioned_task_dag"
+
+        with dag_maker(dag_id, session=session, bundle_version="commit-one"):
+            EmptyOperator(task_id=self.task_id, retries=4)
+        session.commit()
+        first_version_number = DagVersion.get_version(dag_id, session=session).version_number
+
+        with dag_maker(dag_id, session=session, bundle_version="commit-two"):
+            EmptyOperator(task_id=self.task_id, retries=0)
+        session.commit()
+
+        assert DagVersion.get_version(dag_id, session=session).version_number != first_version_number
+
+        url = f"{self.api_prefix}/{dag_id}/tasks/{self.task_id}"
+
+        latest = test_client.get(url)
+        assert latest.status_code == 200
+        assert latest.json()["retries"] == 0
+
+        pinned = test_client.get(url, params={"version_number": first_version_number})
+        assert pinned.status_code == 200
+        assert pinned.json()["retries"] == 4
+
+    @pytest.mark.parametrize("version_number", [0, 99], ids=["zero", "never-written"])
+    def test_unknown_version_number_is_not_found(self, test_client, version_number):
+        """Without this, an unresolvable version would reach ``get_task`` and raise a 500."""
+        response = test_client.get(
+            f"{self.api_prefix}/{self.dag_id}/tasks/{self.task_id}",
+            params={"version_number": version_number},
+        )
+        assert response.status_code == 404
+
     def test_should_respond_404(self, test_client):
         task_id = "xxxx_not_existing"
         response = test_client.get(