Fix Dag serialization crash for Task SDK EventsTimetable

airflow-core/newsfragments/69401.bugfix.rst

@@ -0,0 +1 @@
+Fix Dag serialization crash for Dags scheduled with the Task SDK ``EventsTimetable``.

airflow-core/src/airflow/timetables/events.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import contextlib
 import itertools
 from typing import TYPE_CHECKING, Any
 
@@ -137,5 +138,6 @@ def deserialize(cls, data: dict[str, Any]) -> Timetable:
             presorted=True,
             description=data["description"],
         )
-        timetable._summary = data["_summary"]
+        with contextlib.suppress(KeyError):
+            timetable._summary = data["_summary"]
         return timetable

airflow-core/tests/unit/serialization/test_dag_serialization.py

@@ -61,7 +61,17 @@
 from airflow.models.xcom import XCOM_RETURN_KEY, XComModel
 from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
 from airflow.providers.standard.operators.bash import BashOperator
-from airflow.sdk import DAG, Asset, AssetAlias, BaseHook, TaskGroup, WeightRule, XComArg, teardown
+from airflow.sdk import (
+    DAG,
+    Asset,
+    AssetAlias,
+    BaseHook,
+    EventsTimetable,
+    TaskGroup,
+    WeightRule,
+    XComArg,
+    teardown,
+)
 from airflow.sdk.bases.decorator import DecoratedOperator
 from airflow.sdk.bases.operator import OPERATOR_DEFAULTS, BaseOperator
 from airflow.sdk.definitions._internal.expandinput import EXPAND_INPUT_EMPTY
@@ -781,6 +791,28 @@ def test_dag_roundtrip_from_timetable(self, timetable):
         roundtripped = DagSerialization.from_json(DagSerialization.to_json(dag))
         self.validate_deserialized_dag(roundtripped, dag)
 
+    @pytest.mark.db_test
+    @pytest.mark.parametrize(
+        ("description", "expected_summary"),
+        [
+            pytest.param(None, "2 events", id="no-description"),
+            pytest.param("World Cup", "World Cup", id="with-description"),
+        ],
+    )
+    def test_dag_roundtrip_from_sdk_events_timetable(self, description, expected_summary):
+        """Round-trip a Dag scheduled with the SDK ``EventsTimetable``.
+
+        The SDK timetable serializer does not emit ``_summary``, so the core
+        ``EventsTimetable.deserialize`` must tolerate its absence instead of
+        raising ``KeyError``.
+        """
+        event_dates = [pendulum.datetime(2025, 1, 1), pendulum.datetime(2025, 6, 1)]
+        dag = get_timetable_based_simple_dag(
+            EventsTimetable(event_dates=event_dates, description=description)
+        )
+        roundtripped = DagSerialization.from_json(DagSerialization.to_json(dag))
+        assert roundtripped.timetable.summary == expected_summary
+
     def validate_deserialized_dag(self, serialized_dag: SerializedDAG, dag: DAG):
         """
         Verify that all example DAGs work with DAG Serialization by