From 83806b16b1b70a45bf40bf375df025da832c4db2 Mon Sep 17 00:00:00 2001 From: "Jason(Zhe-You) Liu" <68415893+jason810496@users.noreply.github.com> Date: Fri, 5 Jun 2026 21:52:37 +0800 Subject: [PATCH] [v3-2-test] Fix trigger datetime deserialization (#67795) * Fix trigger datetime deserialization * Fix test_encoders unit test (cherry picked from commit 18caee1497217ef2a0dd0b42bd8a05cb690459aa) Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com> --- .../tests/unit/models/test_trigger.py | 21 +++++++++++++++++++ .../airflow_shared/serialization/__init__.py | 1 + .../airflow/sdk/serde/serializers/datetime.py | 6 ++++++ .../tests/task_sdk/serde/test_serializers.py | 18 ++++++++++++++++ 4 files changed, 46 insertions(+) diff --git a/airflow-core/tests/unit/models/test_trigger.py b/airflow-core/tests/unit/models/test_trigger.py index c79a021f8b8d5..5b2cab10642a2 100644 --- a/airflow-core/tests/unit/models/test_trigger.py +++ b/airflow-core/tests/unit/models/test_trigger.py @@ -37,8 +37,10 @@ from airflow.models.xcom import XComModel from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk.definitions.callback import AsyncCallback +from airflow.serialization.encoders import encode_trigger from airflow.serialization.serialized_objects import BaseSerialization from airflow.triggers.base import ( + BaseEventTrigger, BaseTrigger, TaskFailedEvent, TaskSkippedEvent, @@ -921,6 +923,25 @@ def test_kwargs_not_encrypted(): assert trigger.kwargs["param2"] == "value2" +def test_decrypt_kwargs_roundtrips_datetime(): + """ + A datetime kwarg encoded via BaseSerialization (the asset-watcher path) must survive + the encrypt/decrypt round-trip through serde without crashing, and the DAG-side and + DB-side trigger hashes must match so the trigger is not needlessly recreated. + + Regression: serde lacked a legacy-compat mapping for the bare ``datetime`` timestamp, + so ``_decrypt_kwargs`` raised and the asset-watcher trigger could not be read back. + """ + classpath = "airflow.providers.standard.triggers.temporal.DateTimeTrigger" + moment = datetime.datetime(2026, 1, 15, 12, 30, tzinfo=datetime.timezone.utc) + + dag_kwargs = encode_trigger({"classpath": classpath, "kwargs": {"moment": moment}})["kwargs"] + decrypted = Trigger._decrypt_kwargs(Trigger.encrypt_kwargs(dag_kwargs)) + + assert decrypted["moment"].timestamp() == moment.timestamp() + assert BaseEventTrigger.hash(classpath, dag_kwargs) == BaseEventTrigger.hash(classpath, decrypted) + + def test_asset_trigger_unassigned_included(session): """Asset triggers with triggerer_id=None are returned.""" asset = AssetModel("test_asset") diff --git a/shared/serialization/src/airflow_shared/serialization/__init__.py b/shared/serialization/src/airflow_shared/serialization/__init__.py index 824848dc679b7..03c00157ef218 100644 --- a/shared/serialization/src/airflow_shared/serialization/__init__.py +++ b/shared/serialization/src/airflow_shared/serialization/__init__.py @@ -47,4 +47,5 @@ "set": "builtins.set", "frozenset": "builtins.frozenset", "timedelta": "datetime.timedelta", + "datetime": "datetime.datetime", } diff --git a/task-sdk/src/airflow/sdk/serde/serializers/datetime.py b/task-sdk/src/airflow/sdk/serde/serializers/datetime.py index 9c21ee115a21b..4b471c56ef756 100644 --- a/task-sdk/src/airflow/sdk/serde/serializers/datetime.py +++ b/task-sdk/src/airflow/sdk/serde/serializers/datetime.py @@ -95,6 +95,12 @@ def deserialize(cls: type, version: int, data: dict | str) -> datetime.date | da if cls is datetime.datetime and isinstance(data, dict): return datetime.datetime.fromtimestamp(float(data[TIMESTAMP]), tz=tz) + if cls is datetime.datetime and isinstance(data, int | float): + # Legacy BaseSerialization stored datetimes as a bare UTC timestamp float + # (rather than serde's {timestamp, tz} dict). Round-trip that form so trigger + # kwargs encoded via BaseSerialization can be read back through serde. + return datetime.datetime.fromtimestamp(float(data), tz=datetime.timezone.utc) + if cls is DateTime and isinstance(data, dict): return DateTime.fromtimestamp(float(data[TIMESTAMP]), tz=tz) diff --git a/task-sdk/tests/task_sdk/serde/test_serializers.py b/task-sdk/tests/task_sdk/serde/test_serializers.py index 7935ee2e4c6e5..7e2ade64ebdcf 100644 --- a/task-sdk/tests/task_sdk/serde/test_serializers.py +++ b/task-sdk/tests/task_sdk/serde/test_serializers.py @@ -146,6 +146,24 @@ def test_deserialize_datetime_v1(self, tz_input, expected_tz_name): if tz_input in ["EDT", "CDT", "MDT", "PDT"]: assert deserialize(serialize(deserialized_dt)) == deserialized_dt + def test_deserialize_legacy_datetime_bare_timestamp(self): + """Legacy BaseSerialization stored datetimes as ``{"__type": "datetime", "__var": }``. + + serde must read that form back (used e.g. for trigger kwargs encoded via + BaseSerialization) and reconstruct a UTC ``datetime`` with the same instant. + """ + moment = datetime.datetime(2026, 1, 15, 12, 30, tzinfo=datetime.timezone.utc) + legacy = {"__type": "datetime", "__var": moment.timestamp()} + + deserialized = deserialize(legacy) + + assert isinstance(deserialized, datetime.datetime) + assert deserialized.timestamp() == moment.timestamp() + + # The same form nested inside a dict (the shape trigger kwargs take). + nested = deserialize({"moment": legacy}) + assert nested["moment"].timestamp() == moment.timestamp() + @pytest.mark.parametrize( ("expr", "expected"), [("1", "1"), ("52e4", "520000"), ("2e0", "2"), ("12e-2", "0.12"), ("12.34", "12.34")],