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 e9d1066)

Co-authored-by: Richard Wu <richard9@ualberta.ca>
Co-authored-by: D. Ferruzzi <ferruzzi@amazon.com>

Author: 3 people

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):

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -2278,16 +2278,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

airflow-core/src/airflow/models/serialized_dag.py

@@ -431,7 +431,7 @@ def _try_reuse_deadline_uuids(
         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 @@ def _try_reuse_deadline_uuids(
         :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 @@ def _definitions_match(deadline_data: dict, existing: DeadlineAlertModel) -> boo
 
         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 @@ def _definitions_match(deadline_data: dict, existing: DeadlineAlertModel) -> boo
 
                 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 @@ def _definitions_match(deadline_data: dict, existing: DeadlineAlertModel) -> boo
                 # 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 @@ def _create_deadline_alert_records(
         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 @@ def write_dag(
         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 @@ def write_dag(
                 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 @@ def write_dag(
             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
 

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),
     )
 
 

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

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),

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

@@ -8121,18 +8121,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'
@@ -8141,7 +8142,7 @@ export const $DeadlineResponse = {
                     type: 'null'
                 }
             ],
-            title: 'Alert Description'
+            title: 'Alert Name'
         }
     },
     type: 'object',

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

@@ -2008,8 +2008,8 @@ export type DeadlineResponse = {
     deadline_time: string;
     missed: boolean;
     created_at: string;
+    alert_id?: string | null;
     alert_name?: string | null;
-    alert_description?: string | null;
 };
 
 /**

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py

@@ -52,7 +52,6 @@
 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 @@ def test_single_deadline_without_alert(self, test_client):
         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 @@ def test_missed_deadline_is_reflected(self, test_client):
         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):

airflow-core/tests/unit/models/test_deadline_alert.py

@@ -34,7 +34,6 @@
 
 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 @@ def test_deadline_alert_creation(self, deadline_alert_orm):
         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 @@ def test_minimal_deadline_alert_creation(self, dag_maker, session, deadline_refe
             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)