You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/04/23 05:04:14 UTC
(airflow) 34/48: Add name fields to SDK deadline alerts (#64926) (#65601)
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cfa3d30e0e7818604d02877dbbd4a8fd04db0781
Author: Pierre Jeambrun <pi...@gmail.com>
AuthorDate: Tue Apr 21 18:06:04 2026 +0200
Add name fields to SDK deadline alerts (#64926) (#65601)
* Add deadlines support with name and description fields in alerts and UI
* Add 'viewAll' label to deadlineStatus in dag.json
* Refactor deadlineAlerts referenceType structure
* Refine deadline alert translations in dag.json
Updated deadline alert messages for clarity and consistency.
* Add completion rule text for deadline alerts in UI
* Remove duplicate completionRule entry in deadlineAlerts
* Remove alert description from DeadlineAlert and related models
* Enhance deadline handling: return name updates alongside UUID mapping in SerializedDagModel
* Add alert_id field to DeadlineResponse and update tests for alert handling
* Remove DEADLINES option from MenuItem enum
* Update airflow-core/src/airflow/serialization/encoders.py
---------
(cherry picked from commit e9d106667c7fe98f77c1312bb082c010b472c388)
Co-authored-by: Richard Wu <ri...@ualberta.ca>
Co-authored-by: D. Ferruzzi <fe...@amazon.com>
---
.../api_fastapi/core_api/datamodels/ui/deadline.py | 4 +-
.../api_fastapi/core_api/openapi/_private_ui.yaml | 9 ++--
airflow-core/src/airflow/models/serialized_dag.py | 39 +++++++++++---
airflow-core/src/airflow/serialization/decoders.py | 1 +
.../airflow/serialization/definitions/deadline.py | 2 +
airflow-core/src/airflow/serialization/encoders.py | 1 +
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 11 ++--
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
.../core_api/routes/ui/test_deadlines.py | 12 ++---
.../tests/unit/models/test_deadline_alert.py | 4 --
.../tests/unit/models/test_serialized_dag.py | 62 ++++++++++++++++++++++
task-sdk/src/airflow/sdk/definitions/deadline.py | 2 +
12 files changed, 118 insertions(+), 31 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
index 1ceddeac318..7d616d0449d 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
@@ -33,10 +33,8 @@ class DeadlineResponse(BaseModel):
deadline_time: datetime
missed: bool
created_at: datetime
+ alert_id: UUID | None = Field(validation_alias="deadline_alert_id", default=None)
alert_name: str | None = Field(validation_alias=AliasPath("deadline_alert", "name"), default=None)
- alert_description: str | None = Field(
- validation_alias=AliasPath("deadline_alert", "description"), default=None
- )
class DeadlineCollectionResponse(BaseModel):
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 87ed72c730b..54d93597fad 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -2154,16 +2154,17 @@ components:
type: string
format: date-time
title: Created At
- alert_name:
+ alert_id:
anyOf:
- type: string
+ format: uuid
- type: 'null'
- title: Alert Name
- alert_description:
+ title: Alert Id
+ alert_name:
anyOf:
- type: string
- type: 'null'
- title: Alert Description
+ title: Alert Name
type: object
required:
- id
diff --git a/airflow-core/src/airflow/models/serialized_dag.py b/airflow-core/src/airflow/models/serialized_dag.py
index 0aaf295a58f..8323c2228ff 100644
--- a/airflow-core/src/airflow/models/serialized_dag.py
+++ b/airflow-core/src/airflow/models/serialized_dag.py
@@ -431,7 +431,7 @@ class SerializedDagModel(Base):
existing_deadline_uuids: list[str],
new_deadline_data: list[dict],
session: Session,
- ) -> dict[str, dict] | None:
+ ) -> tuple[dict[str, dict], dict[str, str | None]] | None:
"""
Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
@@ -440,7 +440,11 @@ class SerializedDagModel(Base):
:param existing_deadline_uuids: List of UUID strings from existing serialized Dag
:param new_deadline_data: List of new deadline alert data dicts from the Dag
:param session: Database session
- :return: UUID mapping dict if all match, None if any mismatch detected
+ :return: Tuple of (uuid_mapping, name_updates) if all definitions match, None if any
+ mismatch detected. ``uuid_mapping`` maps UUID string → new deadline data dict.
+ ``name_updates`` maps UUID string → new name **only** for entries whose name
+ changed relative to the existing DB row, so callers can issue targeted UPDATEs
+ and reliably detect whether any DB write occurred.
"""
# defensive check for old 3.1.x format
if existing_deadline_uuids and not isinstance(existing_deadline_uuids[0], str):
@@ -468,6 +472,7 @@ class SerializedDagModel(Base):
matched_uuids: set[UUID] = set()
uuid_mapping: dict[str, dict] = {}
+ name_updates: dict[str, str | None] = {}
for deadline_alert in new_deadline_data:
deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)
@@ -479,9 +484,13 @@ class SerializedDagModel(Base):
if _definitions_match(deadline_data, existing_alert):
# Found a match, reuse this UUID
- uuid_mapping[str(existing_alert.id)] = deadline_data
+ uuid_str = str(existing_alert.id)
+ uuid_mapping[uuid_str] = deadline_data
matched_uuids.add(existing_alert.id)
found_match = True
+ new_name = deadline_data.get(DeadlineAlertFields.NAME)
+ if new_name != existing_alert.name:
+ name_updates[uuid_str] = new_name
break
if not found_match:
@@ -490,7 +499,7 @@ class SerializedDagModel(Base):
# to another deadline), so partial reuse would risk stale cross-references.
return None
- return uuid_mapping
+ return uuid_mapping, name_updates
@classmethod
def _create_deadline_alert_records(
@@ -510,6 +519,7 @@ class SerializedDagModel(Base):
for uuid_str, deadline_data in uuid_mapping.items():
alert = DeadlineAlertModel(
id=UUID(uuid_str),
+ name=deadline_data.get(DeadlineAlertFields.NAME),
reference=deadline_data[DeadlineAlertFields.REFERENCE],
interval=deadline_data[DeadlineAlertFields.INTERVAL],
callback_def=deadline_data[DeadlineAlertFields.CALLBACK],
@@ -625,6 +635,7 @@ class SerializedDagModel(Base):
serialized_dag_hash = _prefetched.dag_hash
dag_version = _prefetched.dag_version
+ name_updated = False
if dag.data.get("dag", {}).get("deadline"):
# Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
# This preserves the hash and avoids unnecessary SerializedDagModel recreations.
@@ -637,16 +648,24 @@ class SerializedDagModel(Base):
and existing_serialized_dag.data
and (existing_deadline_uuids := existing_serialized_dag.data.get("dag", {}).get("deadline"))
):
- deadline_uuid_mapping = cls._try_reuse_deadline_uuids(
+ reuse_result = cls._try_reuse_deadline_uuids(
existing_deadline_uuids,
dag.data["dag"]["deadline"],
session,
)
- if deadline_uuid_mapping is not None:
+ if reuse_result is not None:
+ deadline_uuid_mapping, name_updates = reuse_result
# All deadlines matched — reuse the UUIDs to preserve hash.
- # Clear the mapping since the alert rows already exist in the DB;
- # no need to delete and recreate identical records.
+ # Only issue UPDATE statements for rows whose name actually changed to
+ # avoid unnecessary writes and to make the return value accurate.
+ for uuid_str, new_name in name_updates.items():
+ session.execute(
+ update(DeadlineAlertModel)
+ .where(DeadlineAlertModel.id == UUID(uuid_str))
+ .values(name=new_name)
+ )
+ name_updated = bool(name_updates)
dag.data["dag"]["deadline"] = existing_deadline_uuids
deadline_uuid_mapping = {}
else:
@@ -665,6 +684,10 @@ class SerializedDagModel(Base):
and dag_version
and dag_version.bundle_name == bundle_name
):
+ if name_updated:
+ # The serialized DAG itself is unchanged, but deadline alert name(s) were
+ # updated in the DB, so report True so callers know a write did occur.
+ return True
log.debug("Serialized DAG (%s) is unchanged. Skipping writing to DB", dag.dag_id)
return False
diff --git a/airflow-core/src/airflow/serialization/decoders.py b/airflow-core/src/airflow/serialization/decoders.py
index a25068443e4..27db6134b13 100644
--- a/airflow-core/src/airflow/serialization/decoders.py
+++ b/airflow-core/src/airflow/serialization/decoders.py
@@ -166,6 +166,7 @@ def decode_deadline_alert(encoded_data: dict):
reference=reference,
interval=datetime.timedelta(seconds=data[DeadlineAlertFields.INTERVAL]),
callback=deserialize(data[DeadlineAlertFields.CALLBACK]),
+ name=data.get(DeadlineAlertFields.NAME),
)
diff --git a/airflow-core/src/airflow/serialization/definitions/deadline.py b/airflow-core/src/airflow/serialization/definitions/deadline.py
index 93af9ef19e7..0be0c849107 100644
--- a/airflow-core/src/airflow/serialization/definitions/deadline.py
+++ b/airflow-core/src/airflow/serialization/definitions/deadline.py
@@ -49,6 +49,7 @@ class DeadlineAlertFields:
serializing DeadlineAlert instances to and from their dictionary representation.
"""
+ NAME = "name"
REFERENCE = "reference"
INTERVAL = "interval"
CALLBACK = "callback"
@@ -367,3 +368,4 @@ class SerializedDeadlineAlert:
reference: SerializedReferenceModels.SerializedBaseDeadlineReference
interval: timedelta
callback: Any
+ name: str | None = None
diff --git a/airflow-core/src/airflow/serialization/encoders.py b/airflow-core/src/airflow/serialization/encoders.py
index 2f30511a1e5..87d7ef1f137 100644
--- a/airflow-core/src/airflow/serialization/encoders.py
+++ b/airflow-core/src/airflow/serialization/encoders.py
@@ -211,6 +211,7 @@ def encode_deadline_alert(d: DeadlineAlert | SerializedDeadlineAlert) -> dict[st
from airflow.sdk.serde import serialize
return {
+ "name": d.name,
"reference": encode_deadline_reference(d.reference),
"interval": d.interval.total_seconds(),
"callback": serialize(d.callback),
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 79959b9ceb1..fc2f9a69ad7 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -8026,18 +8026,19 @@ export const $DeadlineResponse = {
format: 'date-time',
title: 'Created At'
},
- alert_name: {
+ alert_id: {
anyOf: [
{
- type: 'string'
+ type: 'string',
+ format: 'uuid'
},
{
type: 'null'
}
],
- title: 'Alert Name'
+ title: 'Alert Id'
},
- alert_description: {
+ alert_name: {
anyOf: [
{
type: 'string'
@@ -8046,7 +8047,7 @@ export const $DeadlineResponse = {
type: 'null'
}
],
- title: 'Alert Description'
+ title: 'Alert Name'
}
},
type: 'object',
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 c29ee5b71cd..82675050838 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
@@ -1970,8 +1970,8 @@ export type DeadlineResponse = {
deadline_time: string;
missed: boolean;
created_at: string;
+ alert_id?: string | null;
alert_name?: string | null;
- alert_description?: string | null;
};
/**
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
index 4dec594eb64..fb69818fa18 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
@@ -52,7 +52,6 @@ RUN_MULTI = "run_multi" # 3 deadlines added out-of-order (ordering test)
RUN_OTHER = "run_other" # has 1 deadline; used to verify per-run isolation
ALERT_NAME = "SLA Breach Alert"
-ALERT_DESCRIPTION = "Fires when SLA is breached"
_CALLBACK_PATH = "tests.unit.api_fastapi.core_api.routes.ui.test_deadlines._noop_callback"
@@ -154,7 +153,6 @@ def setup(dag_maker, session):
alert = DeadlineAlert(
serialized_dag_id=serialized_dag.id,
name=ALERT_NAME,
- description=ALERT_DESCRIPTION,
reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
interval=3600.0,
callback_def={"path": _CALLBACK_PATH},
@@ -226,7 +224,7 @@ class TestGetDagRunDeadlines:
assert deadline1["deadline_time"] == "2025-01-01T12:00:00Z"
assert deadline1["missed"] is False
assert deadline1["alert_name"] is None
- assert deadline1["alert_description"] is None
+ assert deadline1["alert_id"] is None
assert "id" in deadline1
assert "created_at" in deadline1
@@ -237,14 +235,16 @@ class TestGetDagRunDeadlines:
assert data["total_entries"] == 1
assert data["deadlines"][0]["missed"] is True
- def test_deadline_with_alert_name_and_description(self, test_client):
+ def test_deadline_with_alert_name(self, test_client, session):
+ alert = session.scalar(select(DeadlineAlert).where(DeadlineAlert.name == ALERT_NAME))
with assert_queries_count(4):
response = test_client.get(f"/dags/{DAG_ID}/dagRuns/{RUN_ALERT}/deadlines")
assert response.status_code == 200
data = response.json()
assert data["total_entries"] == 1
- assert data["deadlines"][0]["alert_name"] == ALERT_NAME
- assert data["deadlines"][0]["alert_description"] == ALERT_DESCRIPTION
+ dl = data["deadlines"][0]
+ assert dl["alert_name"] == ALERT_NAME
+ assert dl["alert_id"] == str(alert.id)
def test_deadlines_ordered_by_deadline_time_ascending(self, test_client):
with assert_queries_count(4):
diff --git a/airflow-core/tests/unit/models/test_deadline_alert.py b/airflow-core/tests/unit/models/test_deadline_alert.py
index 9d69577a6d6..a9b1854f6ab 100644
--- a/airflow-core/tests/unit/models/test_deadline_alert.py
+++ b/airflow-core/tests/unit/models/test_deadline_alert.py
@@ -34,7 +34,6 @@ from unit.models import DEFAULT_DATE
DAG_ID = "test_deadline_alert_dag"
DEADLINE_NAME = "Test Alert"
-DEADLINE_DESCRIPTION = "This is a test alert description"
DEADLINE_INTERVAL = 60
DEADLINE_CALLBACK = {"path": "test.callback"}
SERIALIZED_DAG_ID = "serialized_dag_uuid"
@@ -62,7 +61,6 @@ def deadline_alert_orm(dag_maker, session, deadline_reference):
alert = DeadlineAlert(
serialized_dag_id=serialized_dag.id,
name=DEADLINE_NAME,
- description=DEADLINE_DESCRIPTION,
reference=deadline_reference,
interval=DEADLINE_INTERVAL,
callback_def=DEADLINE_CALLBACK,
@@ -86,7 +84,6 @@ class TestDeadlineAlert:
assert deadline_alert_orm.id is not None
assert deadline_alert_orm.created_at == DEFAULT_DATE
assert deadline_alert_orm.name == DEADLINE_NAME
- assert deadline_alert_orm.description == DEADLINE_DESCRIPTION
def test_minimal_deadline_alert_creation(self, dag_maker, session, deadline_reference):
with dag_maker(DAG_ID, session=session):
@@ -109,7 +106,6 @@ class TestDeadlineAlert:
assert deadline_alert.id is not None
assert deadline_alert.created_at == DEFAULT_DATE
assert deadline_alert.name is None
- assert deadline_alert.description is None
def test_deadline_alert_repr(self, deadline_alert_orm, deadline_reference):
repr_str = repr(deadline_alert_orm)
diff --git a/airflow-core/tests/unit/models/test_serialized_dag.py b/airflow-core/tests/unit/models/test_serialized_dag.py
index 2185635590e..54438f8e82f 100644
--- a/airflow-core/tests/unit/models/test_serialized_dag.py
+++ b/airflow-core/tests/unit/models/test_serialized_dag.py
@@ -850,3 +850,65 @@ class TestSerializedDagModel:
assert new_serdag_count == 2
assert new_serdag.dag_hash != orig_serdag.dag_hash
assert new_alert.interval == 600.0
+
+ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bundle, session):
+ """Name-only deadline change: UUID reused, DB row updated, write_dag returns True."""
+ dag_id = "test_deadline_name_change"
+
+ dag = DAG(
+ dag_id=dag_id,
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ name="original name",
+ ),
+ )
+ EmptyOperator(task_id="task1", dag=dag)
+ scheduler_dag = sync_dag_to_db(dag, session=session)
+ scheduler_dag.create_dagrun(
+ run_id="test1",
+ run_after=DEFAULT_DATE,
+ state=DagRunState.QUEUED,
+ logical_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ triggered_by=DagRunTriggeredByType.TEST,
+ run_type=DagRunType.MANUAL,
+ )
+ session.commit()
+
+ orig_serdag = session.scalar(select(SDM).where(SDM.dag_id == dag_id).order_by(SDM.created_at.desc()))
+ orig_hash = orig_serdag.dag_hash
+ orig_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id == orig_serdag.id))
+ orig_uuid = orig_alert.id
+
+ # Change only the name — reference, interval, and callback are identical.
+ dag.deadline = DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ name="updated name",
+ )
+
+ did_write = SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing", session=session)
+ session.commit()
+
+ # write_dag must report True because a DB write (name UPDATE) did occur.
+ assert did_write is True
+
+ serdag_count = session.scalar(select(func.count()).select_from(SDM).where(SDM.dag_id == dag_id))
+ latest_serdag = session.scalar(
+ select(SDM).where(SDM.dag_id == dag_id).order_by(SDM.created_at.desc())
+ )
+ updated_alert = session.scalar(select(DAM).where(DAM.id == orig_uuid))
+
+ # No new SerializedDagModel row — UUID was reused so the hash is unchanged.
+ assert serdag_count == 1
+ assert latest_serdag.dag_hash == orig_hash
+
+ # The DeadlineAlert row must still use the same UUID.
+ assert updated_alert is not None
+ assert updated_alert.id == orig_uuid
+
+ # The name must have been updated in the DB.
+ assert updated_alert.name == "updated name"
diff --git a/task-sdk/src/airflow/sdk/definitions/deadline.py b/task-sdk/src/airflow/sdk/definitions/deadline.py
index 2fe220e789d..7164eeec076 100644
--- a/task-sdk/src/airflow/sdk/definitions/deadline.py
+++ b/task-sdk/src/airflow/sdk/definitions/deadline.py
@@ -145,9 +145,11 @@ class DeadlineAlert:
reference: DeadlineReferenceType,
interval: timedelta,
callback: Callback,
+ name: str | None = None,
):
self.reference = reference
self.interval = interval
+ self.name = name
if not isinstance(callback, (AsyncCallback, SyncCallback)):
raise ValueError(f"Callbacks of type {type(callback).__name__} are not currently supported")