This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 5a6f07f36d2 [v3-4-test] Apply note before state in bulk Dag run update
so listeners see it (#74890) (#76523)
5a6f07f36d2 is described below
commit 5a6f07f36d26d77c2a94e8f70fff261664470fe8
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 08:07:05 2026 +0000
[v3-4-test] Apply note before state in bulk Dag run update so listeners see
it (#74890) (#76523)
Dag run state-change listeners fire inside patch_dag_run_state(), and the
bulk update set the note afterwards, so listeners saw the old note. The
single-run PATCH route was fixed the same way in #69873.
(cherry picked from commit 7ee853093ecf1f3c5aba400ee90077e63566e66f)
Co-authored-by: Sahil Ray <[email protected]>
---
.../core_api/services/public/dag_run.py | 4 +-
.../core_api/routes/public/test_dag_run.py | 75 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py
index 291d220a56d..617ed1c86bb 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py
@@ -488,9 +488,11 @@ class BulkDagRunService(BulkService[BulkDAGRunBody]):
dag_run = dag_run_map[key]
if entity.state is not None:
dag = get_dag_for_run(self.dag_bag, dag_run,
session=self.session)
- patch_dag_run_state(dag=dag, dag_run=dag_run,
state=entity.state, session=self.session)
+ # Apply "note" before "state" so listeners fired inside
patch_dag_run_state() see the updated note.
if entity.note is not None:
patch_dag_run_note(dag_run=dag_run, note=entity.note,
user_id=self.user.get_id())
+ if entity.state is not None:
+ patch_dag_run_state(dag=dag, dag_run=dag_run,
state=entity.state, session=self.session)
results.success.append(f"{dag_id}.{run_id}")
except HTTPException as e:
results.errors.append({"error": f"{e.detail}", "status_code":
e.status_code})
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 9610949dded..d3c5900d4e5 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -26,6 +26,7 @@ from unittest import mock
import pytest
import time_machine
import uuid6
+from fastapi import HTTPException, status
from sqlalchemy import delete, func, insert, select, update
from airflow import plugins_manager
@@ -5279,6 +5280,80 @@ class TestBulkDagRuns:
assert dag_run.note == "bulk note"
assert dag_run.state == DAG1_RUN1_STATE
+ @pytest.mark.parametrize("state", [DagRunState.SUCCESS,
DagRunState.FAILED])
+ @pytest.mark.parametrize(
+ ("existing_note", "new_note", "expected_note"),
+ [
+ pytest.param(None, "new note", "new note", id="new-note"),
+ pytest.param("old note", "new note", "new note",
id="replace-stale-note"),
+ pytest.param("old note", "", None, id="empty-string-clears-note"),
+ ],
+ )
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_bulk_update_listener_sees_note_when_note_and_state_both_updated(
+ self, test_client, listener_manager, session, state, existing_note,
new_note, expected_note
+ ):
+ dag_run = session.scalar(
+ select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
DAG1_RUN2_ID)
+ )
+ dag_run.state = DagRunState.RUNNING
+ if existing_note is not None:
+ dag_run.note = (existing_note, "not_test")
+ session.commit()
+ listener = ClassBasedListener()
+ listener_manager(listener)
+ response = test_client.patch(
+ self.ENDPOINT_URL,
+ json={
+ "actions": [
+ {
+ "action": "update",
+ "entities": [{"dag_run_id": DAG1_RUN2_ID, "state":
state, "note": new_note}],
+ }
+ ]
+ },
+ )
+ assert response.status_code == 200
+ body = response.json()
+ assert body["update"]["success"] == [f"{DAG1_ID}.{DAG1_RUN2_ID}"]
+ assert body["update"]["errors"] == []
+ assert listener.state == [state]
+ assert listener.dag_run_note_at_listener == expected_note
+
+
@mock.patch("airflow.api_fastapi.core_api.services.public.dag_run.get_dag_for_run",
autospec=True)
+ def test_bulk_update_dag_not_found_keeps_note_and_state(self,
mock_get_dag_for_run, test_client, session):
+ mock_get_dag_for_run.side_effect = HTTPException(
+ status.HTTP_404_NOT_FOUND, f"The Dag with ID: `{DAG1_ID}` was not
found"
+ )
+ dag_run = session.scalar(
+ select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
DAG1_RUN2_ID)
+ )
+ dag_run.note = ("old note", "not_test")
+ session.commit()
+ response = test_client.patch(
+ self.ENDPOINT_URL,
+ json={
+ "actions": [
+ {
+ "action": "update",
+ "entities": [{"dag_run_id": DAG1_RUN2_ID, "state":
"success", "note": "new note"}],
+ }
+ ]
+ },
+ )
+ assert response.status_code == 200
+ body = response.json()
+ assert body["update"]["success"] == []
+ assert body["update"]["errors"] == [
+ {"error": f"The Dag with ID: `{DAG1_ID}` was not found",
"status_code": 404}
+ ]
+ session.expire_all()
+ dag_run = session.scalar(
+ select(DagRun).where(DagRun.dag_id == DAG1_ID, DagRun.run_id ==
DAG1_RUN2_ID)
+ )
+ assert dag_run.note == "old note"
+ assert dag_run.state == DAG1_RUN2_STATE
+
def test_bulk_update_not_found_fails(self, test_client, session):
"""FAIL semantics: a single missing run fails the whole action and
nothing is updated."""
response = test_client.patch(
