Skip to content

Commit 238dc3d

Browse files
pierrejeambrunpierrejeambrun
committed
UI: Add bulk Clear on the Dag Runs list page
Re-introduces collective Clear on the Dag Runs list page — the Airflow 2.x ``DagRunModelView`` action that the Airflow 3 UI did not yet replicate (apache#63854). The button sits next to the bulk Delete shipped in apache#67095 and opens a dialog mirroring the existing single-run Clear: a segmented control (``Clear existing tasks`` / ``Clear only failed tasks`` / ``Queue up new tasks``), an affected-tasks preview grouped by run, and an optional note. No backend change is required — the dialog fans out the existing ``POST /dags/{dag_id}/dagRuns/{dag_run_id}/clear`` endpoint over the selected runs with ``Promise.allSettled``, then patches the note via ``PATCH /dags/{dag_id}/dagRuns/{dag_run_id}`` on the runs that succeeded. Per-run outcomes are surfaced via the partial-failure UX landed in apache#67284: successful rows are deselected, failures stay in the selection and appear as inline errors so the user can retry just the remaining set. Bulk Mark as success / failed on Dag Runs (the other half of apache#63854) is intentionally out of scope here.
1 parent e56da19 commit 238dc3d

4 files changed

Lines changed: 394 additions & 0 deletions

File tree

‎airflow-core/src/airflow/ui/src/pages/DagRuns/BulkClearDagRunsButton.tsx‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*!
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
import { Button, Flex, Heading, VStack, useDisclosure } from "@chakra-ui/react";
20+
import { useState } from "react";
21+
import { useTranslation } from "react-i18next";
22+
import { CgRedo } from "react-icons/cg";
23+
24+
import type { DAGRunResponse } from "openapi/requests/types.gen";
25+
import { ActionAccordion } from "src/components/ActionAccordion";
26+
import { ActionErrors } from "src/components/ActionErrors";
27+
import { Dialog } from "src/components/ui";
28+
import SegmentedControl from "src/components/ui/SegmentedControl";
29+
import { useBulkClearDagRuns } from "src/queries/useBulkClearDagRuns";
30+
import { useBulkClearDagRunsDryRun } from "src/queries/useBulkClearDagRunsDryRun";
31+
32+
type Props = {
33+
readonly deselectKeys: (keys: Array<string>) => void;
34+
readonly selectedDagRuns: Array<DAGRunResponse>;
35+
};
36+
37+
const BulkClearDagRunsButton = ({ deselectKeys, selectedDagRuns }: Props) => {
38+
const { t: translate } = useTranslation(["common", "dags"]);
39+
const { onClose, onOpen, open } = useDisclosure();
40+
const [selectedOptions, setSelectedOptions] = useState<Array<string>>(["existingTasks"]);
41+
const [note, setNote] = useState<string | null>(null);
42+
const { bulkClear, data, isPending } = useBulkClearDagRuns({
43+
deselectKeys,
44+
onSuccessConfirm: onClose,
45+
});
46+
47+
const handleClose = () => {
48+
setNote(null);
49+
onClose();
50+
};
51+
52+
const onlyFailed = selectedOptions.includes("onlyFailed");
53+
const onlyNew = selectedOptions.includes("newTasks");
54+
55+
const { data: affectedTasks, isFetching } = useBulkClearDagRunsDryRun(open, selectedDagRuns, {
56+
onlyFailed,
57+
onlyNew,
58+
});
59+
60+
return (
61+
<>
62+
<Button onClick={onOpen} size="sm" variant="outline">
63+
<CgRedo />
64+
{translate("dags:runAndTaskActions.clear.button", { type: translate("dagRun_other") })}
65+
</Button>
66+
67+
<Dialog.Root onOpenChange={handleClose} open={open} size="xl">
68+
<Dialog.Content backdrop>
69+
<Dialog.Header>
70+
<VStack align="start" gap={4}>
71+
<Heading size="xl">
72+
{translate("dags:runAndTaskActions.clear.title", { type: translate("dagRun_other") })}
73+
</Heading>
74+
</VStack>
75+
</Dialog.Header>
76+
77+
<Dialog.CloseTrigger />
78+
<Dialog.Body width="full">
79+
<Flex justifyContent="center" mb={4}>
80+
<SegmentedControl
81+
defaultValues={["existingTasks"]}
82+
onChange={setSelectedOptions}
83+
options={[
84+
{
85+
label: translate("dags:runAndTaskActions.options.existingTasks"),
86+
value: "existingTasks",
87+
},
88+
{
89+
label: translate("dags:runAndTaskActions.options.onlyFailed"),
90+
value: "onlyFailed",
91+
},
92+
{
93+
label: translate("dags:runAndTaskActions.options.queueNew"),
94+
value: "newTasks",
95+
},
96+
]}
97+
/>
98+
</Flex>
99+
<ActionAccordion affectedTasks={affectedTasks} groupByRunId note={note} setNote={setNote} />
100+
<ActionErrors actionResponse={data?.clear} error={undefined} />
101+
<Flex justifyContent="end" mt={3}>
102+
<Button
103+
disabled={affectedTasks.total_entries === 0}
104+
loading={isPending || isFetching}
105+
onClick={() => {
106+
void bulkClear(selectedDagRuns, { note, onlyFailed, onlyNew });
107+
}}
108+
>
109+
<CgRedo />
110+
{translate("modal.confirm")}
111+
</Button>
112+
</Flex>
113+
</Dialog.Body>
114+
</Dialog.Content>
115+
</Dialog.Root>
116+
</>
117+
);
118+
};
119+
120+
export default BulkClearDagRunsButton;

‎airflow-core/src/airflow/ui/src/pages/DagRuns/DagRuns.tsx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ import { SearchParamsKeys, type SearchParamsKeysType } from "src/constants/searc
4444
import { useAdvancedSearchArg } from "src/hooks/useAdvancedSearch";
4545
import { renderDuration, useAutoRefresh, isStatePending } from "src/utils";
4646

47+
import BulkClearDagRunsButton from "./BulkClearDagRunsButton";
4748
import BulkDeleteDagRunsButton from "./BulkDeleteDagRunsButton";
4849
import { DagRunsFilters } from "./DagRunsFilters";
4950
import DeleteRunButton from "./DeleteRunButton";
@@ -373,6 +374,7 @@ export const DagRuns = () => {
373374
{selectedRows.size} {translate("selected")}
374375
</ActionBar.SelectionTrigger>
375376
<ActionBar.Separator />
377+
<BulkClearDagRunsButton deselectKeys={deselectKeys} selectedDagRuns={selectedDagRuns} />
376378
<BulkDeleteDagRunsButton deselectKeys={deselectKeys} selectedDagRuns={selectedDagRuns} />
377379
<ActionBar.CloseTrigger onClick={clearSelections} />
378380
</ActionBar.Content>

‎airflow-core/src/airflow/ui/src/queries/useBulkClearDagRuns.ts‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
/*!
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
import { useQueryClient } from "@tanstack/react-query";
20+
import { useState } from "react";
21+
import { useTranslation } from "react-i18next";
22+
23+
import {
24+
UseDagRunServiceGetDagRunKeyFn,
25+
useDagRunServiceGetDagRunsKey,
26+
UseGanttServiceGetGanttDataKeyFn,
27+
useTaskInstanceServiceGetMappedTaskInstanceKey,
28+
useTaskInstanceServiceGetTaskInstanceKey,
29+
useTaskInstanceServiceGetTaskInstancesKey,
30+
} from "openapi/queries";
31+
import { DagRunService } from "openapi/requests/services.gen";
32+
import type { BulkActionResponse, DAGRunResponse } from "openapi/requests/types.gen";
33+
import { toaster } from "src/components/ui";
34+
35+
import { gridQueryKeys, tiPerAttemptQueryKeys } from "./gridViewQueryKeys";
36+
import { useBulkClearDagRunsDryRunKey } from "./useBulkClearDagRunsDryRun";
37+
import { useClearDagRunDryRunKey } from "./useClearDagRunDryRun";
38+
39+
type Props = {
40+
readonly deselectKeys: (keys: Array<string>) => void;
41+
readonly onSuccessConfirm: VoidFunction;
42+
};
43+
44+
export type BulkClearDagRunsOptions = {
45+
note: string | null;
46+
onlyFailed: boolean;
47+
onlyNew: boolean;
48+
};
49+
50+
// Mirrors the bulk-endpoint success key (``{dag_id}.{run_id}``) so callers can pass
51+
// the result straight into ``deselectKeys`` without an extra mapping.
52+
const getRowKey = (dagRun: DAGRunResponse) => `${dagRun.dag_id}.${dagRun.dag_run_id}`;
53+
54+
const formatError = (reason: unknown): string => {
55+
if (reason instanceof Error) {
56+
return reason.message;
57+
}
58+
if (typeof reason === "object" && reason !== null && "body" in reason) {
59+
const { body } = reason as { body?: { detail?: unknown } };
60+
61+
if (body?.detail !== undefined) {
62+
return typeof body.detail === "string" ? body.detail : JSON.stringify(body.detail);
63+
}
64+
}
65+
66+
return String(reason);
67+
};
68+
69+
export const useBulkClearDagRuns = ({ deselectKeys, onSuccessConfirm }: Props) => {
70+
const queryClient = useQueryClient();
71+
const [data, setData] = useState<{ clear: BulkActionResponse } | undefined>(undefined);
72+
const [isPending, setIsPending] = useState(false);
73+
const { t: translate } = useTranslation(["common", "dags"]);
74+
75+
const reset = () => {
76+
setData(undefined);
77+
};
78+
79+
const invalidateQueries = async (dagRuns: ReadonlyArray<DAGRunResponse>) => {
80+
const dagIds = new Set(dagRuns.map((dagRun) => dagRun.dag_id));
81+
const keys = [
82+
[useDagRunServiceGetDagRunsKey],
83+
[useTaskInstanceServiceGetTaskInstancesKey],
84+
[useTaskInstanceServiceGetTaskInstanceKey],
85+
[useTaskInstanceServiceGetMappedTaskInstanceKey],
86+
[useBulkClearDagRunsDryRunKey],
87+
...tiPerAttemptQueryKeys,
88+
...[...dagIds].flatMap((dagId) => [...gridQueryKeys(dagId), [useClearDagRunDryRunKey, dagId]]),
89+
...dagRuns.flatMap((dagRun) => [
90+
UseDagRunServiceGetDagRunKeyFn({ dagId: dagRun.dag_id, dagRunId: dagRun.dag_run_id }),
91+
UseGanttServiceGetGanttDataKeyFn({ dagId: dagRun.dag_id, runId: dagRun.dag_run_id }),
92+
]),
93+
];
94+
95+
await Promise.all(keys.map((queryKey) => queryClient.invalidateQueries({ queryKey })));
96+
};
97+
98+
const bulkClear = async (dagRuns: Array<DAGRunResponse>, options: BulkClearDagRunsOptions) => {
99+
reset();
100+
setIsPending(true);
101+
102+
const settled = await Promise.allSettled(
103+
dagRuns.map((dagRun) =>
104+
DagRunService.clearDagRun({
105+
dagId: dagRun.dag_id,
106+
dagRunId: dagRun.dag_run_id,
107+
requestBody: {
108+
dry_run: false,
109+
only_failed: options.onlyFailed,
110+
only_new: options.onlyNew,
111+
},
112+
}).then(() => dagRun),
113+
),
114+
);
115+
116+
const succeeded: Array<DAGRunResponse> = [];
117+
const errors: Array<Record<string, unknown>> = [];
118+
119+
settled.forEach((outcome, index) => {
120+
if (outcome.status === "fulfilled") {
121+
succeeded.push(outcome.value);
122+
} else {
123+
const dagRun = dagRuns[index];
124+
125+
errors.push({
126+
error: dagRun
127+
? `${getRowKey(dagRun)}: ${formatError(outcome.reason)}`
128+
: formatError(outcome.reason),
129+
});
130+
}
131+
});
132+
133+
if (succeeded.length > 0 && options.note !== null) {
134+
const noteSettled = await Promise.allSettled(
135+
succeeded
136+
.filter((dagRun) => dagRun.note !== options.note)
137+
.map((dagRun) =>
138+
DagRunService.patchDagRun({
139+
dagId: dagRun.dag_id,
140+
dagRunId: dagRun.dag_run_id,
141+
requestBody: { note: options.note },
142+
}).then(() => dagRun),
143+
),
144+
);
145+
146+
noteSettled.forEach((outcome) => {
147+
if (outcome.status === "rejected") {
148+
errors.push({ error: `note: ${formatError(outcome.reason)}` });
149+
}
150+
});
151+
}
152+
153+
await invalidateQueries(dagRuns);
154+
155+
if (succeeded.length > 0) {
156+
toaster.create({
157+
description: translate("toaster.bulkClear.success.description", {
158+
count: succeeded.length,
159+
keys: succeeded.map((dagRun) => dagRun.dag_run_id).join(", "),
160+
resourceName: translate("dagRun_other"),
161+
}),
162+
title: translate("toaster.bulkClear.success.title", {
163+
resourceName: translate("dagRun_other"),
164+
}),
165+
type: "success",
166+
});
167+
deselectKeys(succeeded.map(getRowKey));
168+
}
169+
170+
setData({ clear: { errors, success: succeeded.map(getRowKey) } });
171+
setIsPending(false);
172+
173+
// Per-run failures keep the dialog open so the user can see what failed;
174+
// the consumer renders ``data.clear.errors``.
175+
if (errors.length === 0) {
176+
onSuccessConfirm();
177+
}
178+
};
179+
180+
return { bulkClear, data, isPending, reset };
181+
};

0 commit comments

Comments
 (0)