You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "pierrejeambrun (via GitHub)" <gi...@apache.org> on 2026/05/18 09:10:34 UTC
[PR] Bulk actions for Dag Runs from the runs list page [airflow]
pierrejeambrun opened a new pull request, #67095:
URL: https://github.com/apache/airflow/pull/67095
Re-introduces the Airflow 2.x \"select many runs, then clear / mark / delete\" workflow that the
Flask-AppBuilder ``DagRunModelView`` exposed. Closes #52439.
## Summary
- **Backend**: new ``PATCH /dags/{dag_id}/dagRuns`` endpoint that mirrors the existing
bulk task instances endpoint (``BulkTaskInstanceService``). The new
``BulkDagRunService`` supports:
- bulk **update** (state to ``queued`` / ``success`` / ``failed`` and/or note)
- bulk **delete** (refuses runs in non-final states, matching the single-run delete)
- per-Dag authorization caching when the wildcard path ``/dags/~/dagRuns`` is used
- ``action_on_non_existence: fail | skip`` semantics
- state transitions go through ``set_dag_run_state_to_{success,failed,queued}``
(the same helpers used by the single-run patch endpoint) and emit the
``on_dag_run_success`` / ``on_dag_run_failed`` listener hooks.
- **UI**: the runs list page (``/dags/<dag_id>/runs`` and the global ``/dag_runs``)
now has row-selection checkboxes and an action bar with three buttons that match the
TaskInstances list page: **Clear**, **Mark as success/failed**, **Delete**. Files
added next to ``DagRuns.tsx``:
- ``BulkClearDagRunsButton.tsx`` (uses a new ``useBulkClearDagRuns`` hook that fans
out to the existing single-run clear endpoint so ``only_failed`` keeps working)
- ``BulkMarkDagRunsAsButton.tsx``
- ``BulkDeleteDagRunsButton.tsx``
- ``bulkDagRunsColumns.tsx`` (shared affected-runs table)
- ``useBulkDagRuns.ts`` (mutation hook calling the new bulk endpoint)
- **Tests**: ``TestBulkDagRuns`` in ``test_dag_run.py`` covers update by state, update
by note, cross-Dag wildcard, deletion, deletion of running runs (409), 404
fail/skip, ``create`` returning a 405-style error in ``BulkResponse``, mixed
update+delete actions, listener-hook emission, and 401/403 responses. All 17
pass against the postgres backend.
- **Generated**: ``v2-rest-api-generated.yaml`` and the generated UI client are
regenerated; ``airflow-ctl``'s ``datamodels/generated.py`` is regenerated via
the pre-commit hook.
## Grid view
In 2.x the Grid view did **not** support multi-select either — bulk actions
lived on the ``/dagrun`` list view. This PR keeps the same scope: the runs list
page is the place to filter runs (e.g. ``?state=failed``) and act on a batch.
Adding multi-select to the grid is out of scope for the parity fix and can be a
separate follow-up if the user need warrants it.
## Test plan
- [x] ``breeze run --backend postgres pytest airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py::TestBulkDagRuns -xvs`` — 17/17 pass
- [x] ``prek run --hook-stage pre-commit`` on changed files passes
- [x] UI ``pnpm lint`` (eslint + tsc) passes
- [ ] Manual end-to-end on the runs list page (select runs → Clear / Mark as / Delete)
- [ ] Manual smoke: cross-Dag wildcard call via API
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.7)
Generated-by: Claude Code (Opus 4.7) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #67095:
URL: https://github.com/apache/airflow/pull/67095#issuecomment-4501807829
<!-- uv-lock-rebase-notice -->
`uv.lock` on `main` just moved via [#67232](https://github.com/apache/airflow/pull/67232) ("[main] Upgrade important CI environment"), commit [`6b64a62`](https://github.com/apache/airflow/commit/6b64a621be4c376bf899294e144e7693ae68148d) and this PR currently conflicts.
Quickest fix:
```bash
git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease
```
_Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future `uv.lock` bumps._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3269521670
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+export type BulkActionError = { error: string; status_code?: number };
Review Comment:
What if we defined this in python and it then came from the openapi spec instead of us manually creating an error type here and then casting it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3275484510
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +101,89 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await self._get_dag_run())
yield "\n"
+
+
+class BulkDagRunService(BulkService[BulkDAGRunBody]):
+ """Service for handling bulk operations on Dag Runs."""
+
+ def __init__(
+ self,
+ session: Session,
+ request: BulkBody[BulkDAGRunBody],
+ dag_id: str,
+ ):
+ super().__init__(session, request)
+ self.dag_id = dag_id
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk create is not supported. Use the trigger Dag Run endpoint instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk update is not supported yet. Use the patch Dag Run endpoint per run instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_delete(
Review Comment:
<img width="1018" height="471" alt="Screenshot 2026-05-20 at 18 19 26" src="https://github.com/user-attachments/assets/29556822-66d1-458d-92ec-2fcd5defe84c" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3281635590
##########
uv.lock:
##########
Review Comment:
Let's rebase and get rid of uv lock changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3281303393
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
Review Comment:
Draft PR there, https://github.com/apache/airflow/pull/67284 only the last commit is relevant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #67095:
URL: https://github.com/apache/airflow/pull/67095#issuecomment-4512421838
### Backport failed to create: airflow-ctl/v0-1-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/26250572594'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>airflow-ctl/v0-1-test</td>
<td><a href="https://github.com/apache/airflow/commit/9288949db28e811bb0fbeb88ebd8d509fb159e61"><img src='https://img.shields.io/badge/Commit-9288949-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker 9288949 airflow-ctl/v0-1-test
```
This should apply the commit to the airflow-ctl/v0-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README.md#how-to-backport-pr-with-cherry-picker-cli).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "choo121600 (via GitHub)" <gi...@apache.org>.
choo121600 commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3275593670
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +101,89 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await self._get_dag_run())
yield "\n"
+
+
+class BulkDagRunService(BulkService[BulkDAGRunBody]):
+ """Service for handling bulk operations on Dag Runs."""
+
+ def __init__(
+ self,
+ session: Session,
+ request: BulkBody[BulkDAGRunBody],
+ dag_id: str,
+ ):
+ super().__init__(session, request)
+ self.dag_id = dag_id
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk create is not supported. Use the trigger Dag Run endpoint instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk update is not supported yet. Use the patch Dag Run endpoint per run instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_delete(
Review Comment:
Cool !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3279915998
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
+ const { t: translate } = useTranslation(["common", "dags"]);
+
+ const onSuccess = async (responseData: BulkResponse) => {
+ await Promise.all([
+ queryClient.invalidateQueries({ queryKey: [useDagRunServiceGetDagRunsKey] }),
+ queryClient.invalidateQueries({ queryKey: [useTaskInstanceServiceGetTaskInstancesKey] }),
+ ]);
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3279664186
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
Review Comment:
That's based on the Bulk task instance implementation. I mirrored that.
We can probably improve both of them in a follow up PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3275637894
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
Review Comment:
We shouldn't have useState for errors. tanstack/query should handle this.
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
+ const { t: translate } = useTranslation(["common", "dags"]);
+
+ const onSuccess = async (responseData: BulkResponse) => {
+ await Promise.all([
+ queryClient.invalidateQueries({ queryKey: [useDagRunServiceGetDagRunsKey] }),
+ queryClient.invalidateQueries({ queryKey: [useTaskInstanceServiceGetTaskInstancesKey] }),
+ ]);
+
+ if (responseData.delete) {
+ handleActionResult(responseData.delete, setError, (count, keys) => {
+ toaster.create({
+ description: translate("toaster.bulkDelete.success.description", {
+ count,
+ keys: keys.join(", "),
+ resourceName: translate("dagRun_other"),
+ }),
+ title: translate("toaster.bulkDelete.success.title", {
+ resourceName: translate("dagRun_other"),
+ }),
+ type: "success",
+ });
+ clearSelections();
+ onSuccessConfirm();
+ });
+ }
+ };
+
+ const onError = (_error: unknown) => {
+ setError(_error);
+ };
+
+ const { isPending, mutate } = useDagRunServiceBulkDagRuns({
Review Comment:
```suggestion
const { error, isPending, mutate } = useDagRunServiceBulkDagRuns({
```
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
+ const { t: translate } = useTranslation(["common", "dags"]);
+
+ const onSuccess = async (responseData: BulkResponse) => {
+ await Promise.all([
+ queryClient.invalidateQueries({ queryKey: [useDagRunServiceGetDagRunsKey] }),
+ queryClient.invalidateQueries({ queryKey: [useTaskInstanceServiceGetTaskInstancesKey] }),
+ ]);
+
+ if (responseData.delete) {
+ handleActionResult(responseData.delete, setError, (count, keys) => {
+ toaster.create({
+ description: translate("toaster.bulkDelete.success.description", {
+ count,
+ keys: keys.join(", "),
+ resourceName: translate("dagRun_other"),
+ }),
+ title: translate("toaster.bulkDelete.success.title", {
+ resourceName: translate("dagRun_other"),
+ }),
+ type: "success",
+ });
+ clearSelections();
+ onSuccessConfirm();
+ });
+ }
+ };
+
+ const onError = (_error: unknown) => {
+ setError(_error);
+ };
+
+ const { isPending, mutate } = useDagRunServiceBulkDagRuns({
+ onError,
+ onSuccess,
+ });
+
+ const bulkAction = (requestBody: BulkBody_BulkDAGRunBody_) => {
+ setError(undefined);
+ mutate({ dagId: "~", requestBody });
+ };
+
+ return { bulkAction, error, isPending, setError };
Review Comment:
And setError shouldn't exist here either.
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
+ const { t: translate } = useTranslation(["common", "dags"]);
+
+ const onSuccess = async (responseData: BulkResponse) => {
+ await Promise.all([
+ queryClient.invalidateQueries({ queryKey: [useDagRunServiceGetDagRunsKey] }),
+ queryClient.invalidateQueries({ queryKey: [useTaskInstanceServiceGetTaskInstancesKey] }),
+ ]);
Review Comment:
Let's make sure we are invalidating all relevant queries:
https://github.com/apache/airflow/pull/67212/changes#diff-a980eab73639ff3a89af58d3421ef79f85606c1015664098823b38afa8ed734f
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3279664186
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useState } from "react";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkActionResponse, BulkBody_BulkDAGRunBody_, BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+const handleActionResult = (
+ actionResult: BulkActionResponse,
+ setError: (error: unknown) => void,
+ onSuccess: (count: number, keys: Array<string>) => void,
+) => {
+ const { errors, success } = actionResult;
+
+ if (Array.isArray(errors) && errors.length > 0) {
+ const apiError = errors[0] as { error: string };
+
+ setError({ body: { detail: apiError.error } });
+ } else if (Array.isArray(success) && success.length > 0) {
+ onSuccess(success.length, success);
+ }
+};
+
+export const useBulkDeleteDagRuns = ({ clearSelections, onSuccessConfirm }: Props) => {
+ const queryClient = useQueryClient();
+ const [error, setError] = useState<unknown>(undefined);
Review Comment:
We do this because bulk endpoints can partially succeed. Answer '200' but with the errors field containing the errors. So we need to handle both. Plain error response status code != 2xx that's handled by react query. And response partially succeeded, `errors` field is not empty and response is 200.
```
class BulkActionResponse(BaseModel):
"""
Serializer for individual bulk action responses.
Represents the outcome of a single bulk operation (create, update, or delete).
The response includes a list of successful keys and any errors encountered during the operation.
This structure helps users understand which key actions succeeded and which failed.
"""
success: list[str] = Field(
default=[], description="A list of unique id/key representing successful operations."
)
errors: list[dict[str, Any]] = Field(
default=[],
description="A list of errors encountered during the operation, each containing details about the issue.",
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3275409253
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +101,89 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await self._get_dag_run())
yield "\n"
+
+
+class BulkDagRunService(BulkService[BulkDAGRunBody]):
+ """Service for handling bulk operations on Dag Runs."""
+
+ def __init__(
+ self,
+ session: Session,
+ request: BulkBody[BulkDAGRunBody],
+ dag_id: str,
+ ):
+ super().__init__(session, request)
+ self.dag_id = dag_id
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk create is not supported. Use the trigger Dag Run endpoint instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk update is not supported yet. Use the patch Dag Run endpoint per run instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_delete(
Review Comment:
Yes we should mimic that behavior, thanks for catching that. I'll push an update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #67095:
URL: https://github.com/apache/airflow/pull/67095
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "choo121600 (via GitHub)" <gi...@apache.org>.
choo121600 commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3274644683
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +101,89 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await self._get_dag_run())
yield "\n"
+
+
+class BulkDagRunService(BulkService[BulkDAGRunBody]):
+ """Service for handling bulk operations on Dag Runs."""
+
+ def __init__(
+ self,
+ session: Session,
+ request: BulkBody[BulkDAGRunBody],
+ dag_id: str,
+ ):
+ super().__init__(session, request)
+ self.dag_id = dag_id
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk create is not supported. Use the trigger Dag Run endpoint instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk update is not supported yet. Use the patch Dag Run endpoint per run instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_delete(
Review Comment:
The PR description says RUNNING should return 409 to match the single-run DELETE behavior, but handle_bulk_delete currently skips the state check, and the test confirms that RUNNING DagRuns are deletable.
https://github.com/apache/airflow/pull/67095/changes#diff-912842095111a7bb5d514acbfdc51d19ecad7ea2b6a6e3882f01c20fa87062adR2599-R2600
Is skipping the state check the intended behavior?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3272674407
##########
airflow-core/src/airflow/ui/src/queries/useBulkDeleteDagRuns.ts:
##########
@@ -0,0 +1,97 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useQueryClient } from "@tanstack/react-query";
+import { useTranslation } from "react-i18next";
+
+import {
+ useDagRunServiceBulkDagRuns,
+ useDagRunServiceGetDagRunsKey,
+ useTaskInstanceServiceGetTaskInstancesKey,
+} from "openapi/queries";
+import type { BulkResponse } from "openapi/requests/types.gen";
+import { toaster } from "src/components/ui";
+
+type Props = {
+ readonly clearSelections: VoidFunction;
+ readonly onSuccessConfirm: VoidFunction;
+};
+
+export type BulkActionError = { error: string; status_code?: number };
Review Comment:
I'll refactor the code, that's not needed for other entities, and shouldn't be needed here either. This will remove it altogether.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add bulk delete Dag Runs [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67095:
URL: https://github.com/apache/airflow/pull/67095#discussion_r3275471361
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +101,89 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await self._get_dag_run())
yield "\n"
+
+
+class BulkDagRunService(BulkService[BulkDAGRunBody]):
+ """Service for handling bulk operations on Dag Runs."""
+
+ def __init__(
+ self,
+ session: Session,
+ request: BulkBody[BulkDAGRunBody],
+ dag_id: str,
+ ):
+ super().__init__(session, request)
+ self.dag_id = dag_id
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk create is not supported. Use the trigger Dag Run endpoint instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[BulkDAGRunBody], results: BulkActionResponse
+ ) -> None:
+ results.errors.append(
+ {
+ "error": "Dag Runs bulk update is not supported yet. Use the patch Dag Run endpoint per run instead.",
+ "status_code": status.HTTP_405_METHOD_NOT_ALLOWED,
+ }
+ )
+
+ def handle_bulk_delete(
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org