ferruzzi commented on code in PR #72651:
URL: https://github.com/apache/airflow/pull/72651#discussion_r3970858750
##########
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:
Are these meant to be local imports? I know Claude loves doing that despite
telling it not to. Also, I'm surprised `ruff` didn't complain about the order,
shouldn't the constants be first?
##########
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:
Smallest of nits, feel free to ignore; this is awkward phrasing, maybe
consider `f"Deadline {field} does not name a class."`? Or maybe split it into
two more accurate messages:
```
if not isinstance(converted, dict):
raise ValueError(f"Deadline {field} is not a serialized object.")
if CLASSNAME not in converted:
raise ValueError(f"Deadline {field} does not name a class.")
```
##########
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:
Nit: return type?
##########
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:
I think callbacks need the same treatment as` _VARIABLE_INTERVAL_CLASSNAMES`
since callbacks were also moved in 3.2? Something like
```
_LEGACY_CALLBACK_MODULE = "airflow.sdk.definitions.deadline"
```
then below build `permitted`:
```
permitted = {
f"{module}.{cls.__qualname__}": cls
for cls in (AsyncCallback, SyncCallback)
for module in (cls.__module__, _LEGACY_CALLBACK_MODULE)
}
```
##########
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"):
+ if optional in data:
+ value = data[optional]
+ if value is not None and not isinstance(value, str):
+ raise ValueError(f"Deadline callback {optional} is not a
string.")
+ fields[optional] = value
+
+ unexpected = set(data) - {"path", "kwargs", "queue", "executor"}
Review Comment:
Actually, is there a reason to list the strings here rather than using
`set(callback_cls.serialized_fields())`? If we can do that, it would also kill
my comment above about the magic strings. This just feels like it's asking for
drift later.
And from a purely defensive coding perspective, shouldn't this be the first
check? "If we got unexpected input, abort immediately. THEN go through each
one and check it" but maybe that's overly paranoid?
##########
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:
Just trying to think this through, maybe this should be an
`ALLOWED_OPTIONALS` constant ? But then maybe "path" and "kwargs" could be
constants too to drop the "magic strings" and it's a rabbit hole... so maybe
not.
--
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]
