potiuk commented on code in PR #72651:
URL: https://github.com/apache/airflow/pull/72651#discussion_r3999398288
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -182,15 +182,125 @@ def decode_deadline_reference(reference_data: dict):
return reference_class.deserialize_reference(reference_data)
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+ {
+ "airflow.sdk.definitions.deadline.VariableInterval",
+
"airflow.serialization.definitions.deadline.SerializedVariableInterval",
+ }
+)
+
+
+def _normalised_payload(encoded: Any, field: str) -> tuple[str, Any]:
+ """
+ Return ``(classname, data)`` for a serde payload, in either encoding.
+
+ ``serde`` accepts a legacy ``{"__type": ..., "__var": ...}`` shape and
rewrites it
+ into the current one *inside* ``deserialize``. Anything inspecting the
payload before
+ that call therefore has to normalise it first, or the legacy spelling
carries no
+ ``__classname__`` at the moment it is looked at and slips past unexamined.
+ """
+ from airflow.sdk.serde import _convert, CLASSNAME, DATA
Review Comment:
Taken as written — split into the two checks with your wording.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -182,15 +182,125 @@ def decode_deadline_reference(reference_data: dict):
return reference_class.deserialize_reference(reference_data)
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+ {
+ "airflow.sdk.definitions.deadline.VariableInterval",
+
"airflow.serialization.definitions.deadline.SerializedVariableInterval",
+ }
+)
+
+
+def _normalised_payload(encoded: Any, field: str) -> tuple[str, Any]:
+ """
+ Return ``(classname, data)`` for a serde payload, in either encoding.
+
+ ``serde`` accepts a legacy ``{"__type": ..., "__var": ...}`` shape and
rewrites it
+ into the current one *inside* ``deserialize``. Anything inspecting the
payload before
+ that call therefore has to normalise it first, or the legacy spelling
carries no
+ ``__classname__`` at the moment it is looked at and slips past unexamined.
+ """
+ from airflow.sdk.serde import _convert, CLASSNAME, DATA
+
+ if not isinstance(encoded, dict):
+ raise ValueError(f"Deadline {field} is not a serialized object.")
+ converted = _convert(encoded)
+ if not isinstance(converted, dict) or CLASSNAME not in converted:
+ raise ValueError(f"Deadline {field} names no class.")
Review Comment:
Added: `-> AsyncCallback | SyncCallback`.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -182,15 +182,125 @@ def decode_deadline_reference(reference_data: dict):
return reference_class.deserialize_reference(reference_data)
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+ {
+ "airflow.sdk.definitions.deadline.VariableInterval",
+
"airflow.serialization.definitions.deadline.SerializedVariableInterval",
+ }
+)
+
+
+def _normalised_payload(encoded: Any, field: str) -> tuple[str, Any]:
+ """
+ Return ``(classname, data)`` for a serde payload, in either encoding.
+
+ ``serde`` accepts a legacy ``{"__type": ..., "__var": ...}`` shape and
rewrites it
+ into the current one *inside* ``deserialize``. Anything inspecting the
payload before
+ that call therefore has to normalise it first, or the legacy spelling
carries no
+ ``__classname__`` at the moment it is looked at and slips past unexamined.
+ """
+ from airflow.sdk.serde import _convert, CLASSNAME, DATA
+
+ if not isinstance(encoded, dict):
+ raise ValueError(f"Deadline {field} is not a serialized object.")
+ converted = _convert(encoded)
+ if not isinstance(converted, dict) or CLASSNAME not in converted:
+ raise ValueError(f"Deadline {field} names no class.")
+ return converted[CLASSNAME], converted.get(DATA)
+
+
+def _decode_deadline_interval(raw_interval: Any) -> datetime.timedelta |
SerializedVariableInterval:
+ """
+ Build the interval from its encoded form without importing what the
payload names.
+
+ Only three shapes are legitimate, and each is reconstructed from
primitives directly.
+ Nothing here reaches ``serde.deserialize``, so no class named by a Dag
author is
+ imported or instantiated in the scheduler or the API server.
+ """
+ # Backward compatibility: previously stored as total_seconds().
+ if isinstance(raw_interval, (int, float)) and not isinstance(raw_interval,
bool):
+ return datetime.timedelta(seconds=raw_interval)
+
+ classname, data = _normalised_payload(raw_interval, "interval")
+
+ if classname == _TIMEDELTA_CLASSNAME:
+ if isinstance(data, (int, float)) and not isinstance(data, bool):
+ return datetime.timedelta(seconds=data)
+ raise ValueError("Deadline interval timedelta payload is not a
number.")
+
+ if classname in _VARIABLE_INTERVAL_CLASSNAMES:
+ key = data.get("key") if isinstance(data, dict) else None
+ if not isinstance(key, str):
+ raise ValueError("Deadline interval variable payload has no string
key.")
+ return SerializedVariableInterval(key=key)
+
+ raise ValueError(
+ f"Refusing to deserialize {classname!r} as a deadline interval. "
+ f"Permitted: {_TIMEDELTA_CLASSNAME}, {',
'.join(sorted(_VARIABLE_INTERVAL_CLASSNAMES))}."
+ )
+
+
+def _decode_deadline_callback(raw_callback: Any):
Review Comment:
Your instinct to not go down the rabbit hole was right, but your next
comment pointed at the better exit — the classes already declare their own
fields, so there's nothing to name a constant for. See below.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -182,15 +182,125 @@ def decode_deadline_reference(reference_data: dict):
return reference_class.deserialize_reference(reference_data)
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+ {
+ "airflow.sdk.definitions.deadline.VariableInterval",
+
"airflow.serialization.definitions.deadline.SerializedVariableInterval",
+ }
+)
+
+
+def _normalised_payload(encoded: Any, field: str) -> tuple[str, Any]:
+ """
+ Return ``(classname, data)`` for a serde payload, in either encoding.
+
+ ``serde`` accepts a legacy ``{"__type": ..., "__var": ...}`` shape and
rewrites it
+ into the current one *inside* ``deserialize``. Anything inspecting the
payload before
+ that call therefore has to normalise it first, or the legacy spelling
carries no
+ ``__classname__`` at the moment it is looked at and slips past unexamined.
+ """
+ from airflow.sdk.serde import _convert, CLASSNAME, DATA
+
+ if not isinstance(encoded, dict):
+ raise ValueError(f"Deadline {field} is not a serialized object.")
+ converted = _convert(encoded)
+ if not isinstance(converted, dict) or CLASSNAME not in converted:
+ raise ValueError(f"Deadline {field} names no class.")
+ return converted[CLASSNAME], converted.get(DATA)
+
+
+def _decode_deadline_interval(raw_interval: Any) -> datetime.timedelta |
SerializedVariableInterval:
+ """
+ Build the interval from its encoded form without importing what the
payload names.
+
+ Only three shapes are legitimate, and each is reconstructed from
primitives directly.
+ Nothing here reaches ``serde.deserialize``, so no class named by a Dag
author is
+ imported or instantiated in the scheduler or the API server.
+ """
+ # Backward compatibility: previously stored as total_seconds().
+ if isinstance(raw_interval, (int, float)) and not isinstance(raw_interval,
bool):
+ return datetime.timedelta(seconds=raw_interval)
+
+ classname, data = _normalised_payload(raw_interval, "interval")
+
+ if classname == _TIMEDELTA_CLASSNAME:
+ if isinstance(data, (int, float)) and not isinstance(data, bool):
+ return datetime.timedelta(seconds=data)
+ raise ValueError("Deadline interval timedelta payload is not a
number.")
+
+ if classname in _VARIABLE_INTERVAL_CLASSNAMES:
+ key = data.get("key") if isinstance(data, dict) else None
+ if not isinstance(key, str):
+ raise ValueError("Deadline interval variable payload has no string
key.")
+ return SerializedVariableInterval(key=key)
+
+ raise ValueError(
+ f"Refusing to deserialize {classname!r} as a deadline interval. "
+ f"Permitted: {_TIMEDELTA_CLASSNAME}, {',
'.join(sorted(_VARIABLE_INTERVAL_CLASSNAMES))}."
+ )
+
+
+def _decode_deadline_callback(raw_callback: Any):
+ """
+ Build the callback from its encoded form without importing what the
payload names.
+
+ ``kwargs`` is still passed through generic deserialization, and that is a
deliberate,
+ documented limit rather than an oversight. Leaving it encoded would close
a real
+ residual -- a legitimate callback can carry an arbitrary allow-listed
class under its
+ kwargs, which serde constructs while the outer payload looks entirely
valid -- but the
+ kwargs are consumed through two different paths that use two different
encodings
+ (``BaseSerialization`` in the triggerer, serde here), and deferring the
decode without
+ getting both exactly right silently hands user code an encoded dict in
place of its
+ argument. That residual is not specific to deadlines: it is the general
property of
+ deserializing Dag-author data, shared with every other serde call site.
Closing it
+ belongs with that broader work, not smuggled in here.
+ """
+ from airflow.sdk.definitions.callback import (
+ AsyncCallback,
+ SyncCallback,
+ _SerializedCallbackPath,
+ )
+
+ permitted = {f"{cls.__module__}.{cls.__qualname__}": cls for cls in
(AsyncCallback, SyncCallback)}
+ classname, data = _normalised_payload(raw_callback, "callback")
+ callback_cls = permitted.get(classname)
+ if callback_cls is None:
+ raise ValueError(
+ f"Refusing to deserialize {classname!r} as a deadline callback. "
+ f"Permitted: {', '.join(sorted(permitted))}."
+ )
+ if not isinstance(data, dict):
+ raise ValueError("Deadline callback payload is not a mapping.")
+
+ path = data.get("path")
+ if not isinstance(path, str):
+ raise ValueError("Deadline callback payload has no string path.")
+
+ from airflow.sdk.serde import deserialize
+
+ raw_kwargs = data.get("kwargs") or {}
+ fields: dict[str, Any] = {"kwargs": deserialize(raw_kwargs) if raw_kwargs
else {}}
+ for optional in ("queue", "executor"):
Review Comment:
Good catch, this was a real bug. `5f62e838e2` (#58177) moved both classes
out of `airflow.sdk.definitions.deadline` into `...definitions.callback`, and
it shipped in **3.2.0** — so alerts serialized by 3.1.x name the old path and
this gate would have made them undecodable on upgrade. Applied your
`_LEGACY_CALLBACK_MODULE` suggestion essentially verbatim, with a test per
subclass that fails without it.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -182,15 +182,125 @@ def decode_deadline_reference(reference_data: dict):
return reference_class.deserialize_reference(reference_data)
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+ {
+ "airflow.sdk.definitions.deadline.VariableInterval",
+
"airflow.serialization.definitions.deadline.SerializedVariableInterval",
+ }
+)
+
Review Comment:
Both right, and the drift you were worried about is already here rather than
hypothetical. `serialized_fields()` is **per class** —
`("path","kwargs","queue")` for `AsyncCallback`, `("path","kwargs","executor")`
for `SyncCallback` — so the hardcoded four-element set accepted `queue` on a
`SyncCallback`, which then reached the constructor and raised `TypeError:
SyncCallback.__init__() got an unexpected keyword argument 'queue'` from inside
the rebuild. Now asks the class, and the unexpected-field check moved ahead of
everything that reconstructs from the payload, as you suggested. Test covers
both directions.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
