Repository: apache/airflow
File: airflow-core/src/airflow/models/serialized_dag.py (Python)

Code before:
            return uuid_mapping

        for deadline_alert in dag_deadline_data:
            deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)

            deadline_uuid = str(uuid6.uuid7())
            uuid_mapping[deadline_uuid] = deadline_data

        dag_data["dag"]["deadline"] = list(uuid_mapping.keys())

        return uuid_mapping

    @classmethod
    def _try_reuse_deadline_uuids(
        cls,
        existing_deadline_uuids: list[str],
        new_deadline_data: list[dict],
        session: Session,
    ) -> dict[str, dict] | None:
        """
        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.

        Returns None if Deadline hashes are not all identical, indicating they need to be updated.

        :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
        """
        if len(existing_deadline_uuids) != len(new_deadline_data):
            return None

        existing_alerts = session.scalars(
            select(DeadlineAlertModel).where(DeadlineAlertModel.id.in_(existing_deadline_uuids))
        ).all()

        if len(existing_alerts) != len(existing_deadline_uuids):
            return None

        new_alerts_temp = []
        for deadline_alert in new_deadline_data:
            deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)
            # Create a temporary alert for comparison
            temp_alert = DeadlineAlertModel(
                id="temp",  # id is required for the object but isn't included in the __eq__
                reference=deadline_data[DeadlineAlertFields.REFERENCE],
                interval=deadline_data[DeadlineAlertFields.INTERVAL],
                callback_def=deadline_data[DeadlineAlertFields.CALLBACK],
            )
            new_alerts_temp.append((temp_alert, deadline_data))


Review comment (suggestion):
```suggestion
        :param new_deadline_data: List of new deadline alert data dicts from the Dag
```

Code after:

        for deadline_alert in dag_deadline_data:
            deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)

            deadline_uuid = str(uuid6.uuid7())
            uuid_mapping[deadline_uuid] = deadline_data

        dag_data["dag"]["deadline"] = list(uuid_mapping.keys())

        return uuid_mapping

    @classmethod
    def _try_reuse_deadline_uuids(
        cls,
        existing_deadline_uuids: list[str],
        new_deadline_data: list[dict],
        session: Session,
    ) -> dict[str, dict] | None:
        """
        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.

        Returns None if Deadline hashes are not all identical, indicating they need to be updated.

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

        def _definitions_match(deadline_data: dict, existing: DeadlineAlertModel) -> bool:
            """Check if raw deadline data matches an existing DeadlineAlert's definition."""
            return (
                deadline_data[DeadlineAlertFields.REFERENCE] == existing.reference
                and deadline_data[DeadlineAlertFields.INTERVAL] == existing.interval
                and deadline_data[DeadlineAlertFields.CALLBACK] == existing.callback_def
            )

        if len(existing_deadline_uuids) != len(new_deadline_data):
            return None

        existing_deadline_uuids_as_uuid = [UUID(uid) for uid in existing_deadline_uuids]
        existing_alerts = session.scalars(
            select(DeadlineAlertModel).where(DeadlineAlertModel.id.in_(existing_deadline_uuids_as_uuid))
        ).all()

        if len(existing_alerts) != len(existing_deadline_uuids):
            return None

        matched_uuids: set[UUID] = set()
        uuid_mapping: dict[str, dict] = {}