This is an automated email from the ASF dual-hosted git repository.
ferruzzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b9e46d64ee5 Add type-checking guards to DeadlineAlert init (#73294)
b9e46d64ee5 is described below
commit b9e46d64ee5231db886e0419bcb6ebe32a4d022f
Author: D. Ferruzzi <[email protected]>
AuthorDate: Tue Sep 22 15:21:16 2026 -0700
Add type-checking guards to DeadlineAlert init (#73294)
* Add type-checking guards to DeadlineAlert init
---
airflow-core/newsfragments/73294.significant.rst | 14 ++
airflow-core/src/airflow/serialization/decoders.py | 5 +-
.../unit/serialization/test_serialized_objects.py | 11 +-
task-sdk/src/airflow/sdk/definitions/callback.py | 6 +
task-sdk/src/airflow/sdk/definitions/deadline.py | 38 +++++-
.../tests/task_sdk/definitions/test_deadline.py | 148 ++++++++++++++++++---
6 files changed, 192 insertions(+), 30 deletions(-)
diff --git a/airflow-core/newsfragments/73294.significant.rst
b/airflow-core/newsfragments/73294.significant.rst
new file mode 100644
index 00000000000..33995056919
--- /dev/null
+++ b/airflow-core/newsfragments/73294.significant.rst
@@ -0,0 +1,14 @@
+Deadline alert callbacks are validated at Dag parse time, and numeric
intervals are deprecated
+
+``DeadlineAlert`` now refuses a ``callback`` that is not exactly
``AsyncCallback`` or
+``SyncCallback``, so a subclass fails when the Dag is parsed rather than when
the scheduler
+reads it back. The scheduler decodes stored callbacks by exact class name so
no class
+named by a Dag author is imported or instantiated there, which means a
subclass could be
+written and then never read.
+
+Passing a number as the ``interval`` is deprecated and now emits
``RemovedInAirflow4Warning``.
+It is still accepted and taken as a count of seconds, but the supported types
are ``timedelta``
+and ``VariableInterval``. A ``timedelta`` subclass such as
``pendulum.duration(...)`` is
+accepted and normalized to a plain ``timedelta``, where previously it passed
validation and
+then failed during serialization. ``interval=True`` is refused, because
``bool`` is an ``int``
+subclass and would otherwise be taken as one second.
diff --git a/airflow-core/src/airflow/serialization/decoders.py
b/airflow-core/src/airflow/serialization/decoders.py
index ca5db04200d..fb1651bfb01 100644
--- a/airflow-core/src/airflow/serialization/decoders.py
+++ b/airflow-core/src/airflow/serialization/decoders.py
@@ -263,14 +263,13 @@ def _decode_deadline_callback(raw_callback: Any) ->
AsyncCallback | SyncCallback
belongs with that broader work, not smuggled in here.
"""
from airflow.sdk.definitions.callback import (
- AsyncCallback,
- SyncCallback,
+ DEADLINE_CALLBACK_TYPES,
_SerializedCallbackPath,
)
permitted = {
f"{module}.{cls.__qualname__}": cls
- for cls in (AsyncCallback, SyncCallback)
+ for cls in DEADLINE_CALLBACK_TYPES
for module in (cls.__module__, _LEGACY_CALLBACK_MODULE)
}
classname, data = _normalised_payload(raw_callback, "callback")
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py
b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 3e0812cfcb9..f1d44c211f0 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -499,30 +499,33 @@ def test_serialize_deserialize_connection():
@pytest.mark.parametrize("reference", REFERENCE_TYPES)
@pytest.mark.parametrize(
- ("interval", "expected_interval"),
+ ("alert_class", "interval", "expected_interval"),
[
pytest.param(
+ DeadlineAlert,
timedelta(hours=1),
timedelta(hours=1),
id="timedelta",
),
pytest.param(
+ DeadlineAlert,
VariableInterval("deadline_seconds"),
SerializedVariableInterval("deadline_seconds"),
id="sdk_variable_interval",
),
pytest.param(
+ SerializedDeadlineAlert,
SerializedVariableInterval("deadline_seconds"),
SerializedVariableInterval("deadline_seconds"),
- id="serialized_variable_interval",
+ id="core_serialized_alert",
),
],
)
-def test_serialize_deserialize_deadline_alert(reference, interval,
expected_interval):
+def test_serialize_deserialize_deadline_alert(reference, alert_class,
interval, expected_interval):
public_deadline_alert_fields = {
field.lower() for field in vars(DeadlineAlertFields) if not
field.startswith("_")
}
- original = DeadlineAlert(
+ original = alert_class(
reference=reference,
interval=interval,
callback=AsyncCallback(empty_callback_for_deadline,
kwargs=TEST_CALLBACK_KWARGS),
diff --git a/task-sdk/src/airflow/sdk/definitions/callback.py
b/task-sdk/src/airflow/sdk/definitions/callback.py
index f12e6343d81..8f48109d63c 100644
--- a/task-sdk/src/airflow/sdk/definitions/callback.py
+++ b/task-sdk/src/airflow/sdk/definitions/callback.py
@@ -199,3 +199,9 @@ class SyncCallback(Callback):
@classmethod
def serialized_fields(cls) -> tuple[str, ...]:
return super().serialized_fields() + ("executor",)
+
+
+# The callback types a ``DeadlineAlert`` may hold. ``DeadlineAlert.__init__``
refuses anything else
+# at Dag-parse time and the scheduler's decoder allow-lists exactly these by
qualified class name, so
+# the two must not drift: both build from this.
+DEADLINE_CALLBACK_TYPES = (AsyncCallback, SyncCallback)
diff --git a/task-sdk/src/airflow/sdk/definitions/deadline.py
b/task-sdk/src/airflow/sdk/definitions/deadline.py
index ba923868b44..3ef17245803 100644
--- a/task-sdk/src/airflow/sdk/definitions/deadline.py
+++ b/task-sdk/src/airflow/sdk/definitions/deadline.py
@@ -25,7 +25,7 @@ from typing import TYPE_CHECKING, Any, overload
import attrs
-from airflow.sdk.definitions.callback import AsyncCallback, Callback,
SyncCallback
+from airflow.sdk.definitions.callback import DEADLINE_CALLBACK_TYPES,
AsyncCallback, SyncCallback
from airflow.sdk.definitions.variable import Variable
from airflow.sdk.exceptions import AirflowRuntimeError,
RemovedInAirflow4Warning
@@ -151,17 +151,43 @@ class DeadlineAlert:
self,
reference: DeadlineReferenceType,
interval: timedelta | VariableInterval,
- callback: Callback,
+ callback: AsyncCallback | SyncCallback,
name: str | None = None,
):
+ if isinstance(interval, (int, float)) and not isinstance(interval,
bool):
+ # A bare number was never documented or type-hinted, but it parses
today because this
+ # check did not exist, and the decoder still reads legacy rows
stored as total_seconds().
+ # Normalize so Dags that parse today keep parsing, and warn so the
accident does not
+ # become contract. bool is excluded: it is an int subclass, so
True would mean 1 second.
+ warnings.warn(
+ f"Passing a number as a deadline interval is deprecated and
will be removed in a "
+ f"future release. Pass timedelta(seconds={interval}) instead.",
+ RemovedInAirflow4Warning,
+ stacklevel=2,
+ )
+ interval = timedelta(seconds=interval)
+ elif isinstance(interval, timedelta):
+ # serde dispatches on qualified class name and registers only
datetime.timedelta, so a
+ # subclass such as pendulum.duration() would pass the check below
and then fail there.
+ # Rebuilding timedelta subclasses into a timedelta keeps this to
one path.
+ interval = timedelta(seconds=interval.total_seconds())
+
+ if not isinstance(interval, (timedelta, VariableInterval)):
+ raise ValueError(
+ f"Interval must be a `timedelta` or a `VariableInterval`,
received {type(interval).__name__}."
+ )
+
+ # Serializing blocks subclasses for security reasons, so isinstance is
too loose.
+ if type(callback) not in DEADLINE_CALLBACK_TYPES:
+ raise ValueError(
+ f"Callbacks must be `AsyncCallback` or `SyncCallback`,
received {type(callback).__name__}."
+ )
+
+ self.callback = callback
self.reference = reference
self.interval = interval
self.name = name
- if not isinstance(callback, (AsyncCallback, SyncCallback)):
- raise ValueError(f"Callbacks of type {type(callback).__name__} are
not currently supported")
- self.callback = callback
-
def __eq__(self, other: object) -> bool:
if not isinstance(other, DeadlineAlert):
return NotImplemented
diff --git a/task-sdk/tests/task_sdk/definitions/test_deadline.py
b/task-sdk/tests/task_sdk/definitions/test_deadline.py
index 47cc6a874d8..a96229420f6 100644
--- a/task-sdk/tests/task_sdk/definitions/test_deadline.py
+++ b/task-sdk/tests/task_sdk/definitions/test_deadline.py
@@ -19,13 +19,14 @@ from __future__ import annotations
from datetime import datetime, timedelta
from unittest import mock
+import pendulum
import pytest
from task_sdk.definitions.test_callback import TEST_CALLBACK_KWARGS,
TEST_CALLBACK_PATH, UNIMPORTABLE_DOT_PATH
from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference,
VariableInterval
from airflow.sdk.definitions.variable import Variable
-from airflow.sdk.exceptions import AirflowRuntimeError
+from airflow.sdk.exceptions import AirflowRuntimeError,
RemovedInAirflow4Warning
DAG_ID = "dag_id_1"
RUN_ID = 1
@@ -43,6 +44,11 @@ TEST_DEADLINE_CALLBACK = AsyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_
class TestDeadlineAlert:
+ class SubclassedCallback(SyncCallback):
+ """Stand-in for a Dag author's own Callback subclass, which
DeadlineAlert must reject."""
+
+ ...
+
@pytest.mark.parametrize(
("test_alert", "should_equal"),
[
@@ -142,29 +148,137 @@ class TestDeadlineAlert:
assert len(alert_set) == 1
@pytest.mark.parametrize(
- ("callback_class"),
+ ("test_callback", "expected_name", "expected_pass"),
[
- pytest.param(AsyncCallback, id="async_callback"),
- pytest.param(SyncCallback, id="sync_callback"),
+ pytest.param(
+ SyncCallback(TEST_CALLBACK_PATH),
+ "SyncCallback",
+ True,
+ id="sync_callback_passes",
+ ),
+ pytest.param(
+ AsyncCallback(TEST_CALLBACK_PATH),
+ "AsyncCallback",
+ True,
+ id="async_callback_passes",
+ ),
+ pytest.param(
+ type(TEST_CALLBACK_PATH),
+ "type",
+ False,
+ id="non_callback_callable_fails",
+ ),
+ pytest.param(
+ SubclassedCallback(TEST_CALLBACK_PATH),
+ "SubclassedCallback",
+ False,
+ id="subclassed_callback_fails",
+ ),
+ pytest.param(
+ "not_a_callback",
+ "str",
+ False,
+ id="non_callback_fails",
+ ),
+ pytest.param(
+ None,
+ "NoneType",
+ False,
+ id="can_not_be_none",
+ ),
],
)
- def test_deadline_alert_accepts_all_callbacks(self, callback_class):
+ def test_deadline_init_callback_type_checks(self, test_callback,
expected_name, expected_pass):
+ if expected_pass:
+ alert = DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(hours=1),
+ callback=test_callback,
+ )
+
+ assert alert.callback is test_callback
+ assert type(alert.callback).__name__ == expected_name
+ else:
+ with pytest.raises(
+ ValueError,
+ match=f"Callbacks must be `AsyncCallback` or `SyncCallback`,
received {expected_name}",
+ ):
+ DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(hours=1),
+ callback=test_callback,
+ )
+
+ @pytest.mark.parametrize(
+ ("test_interval", "expected_name", "expected_pass"),
+ [
+ pytest.param(timedelta(1), "timedelta", True,
id="positive_timedelta_passes"),
+ pytest.param(timedelta(-1), "timedelta", True,
id="negative_timedelta_passes"),
+ pytest.param(timedelta(0), "timedelta", True,
id="zero_timedelta_passes"),
+ pytest.param(VariableInterval("var"), "VariableInterval", True,
id="VariableInterval_passes"),
+ pytest.param(True, "bool", False, id="bool_fails"),
+ pytest.param("str", "str", False, id="string_fails"),
+ pytest.param(None, "NoneType", False, id="can_not_be_none"),
+ ],
+ )
+ def test_deadline_init_interval_type_checks(self, test_interval,
expected_name, expected_pass):
+ if expected_pass:
+ alert = DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=test_interval,
+ callback=TEST_DEADLINE_CALLBACK,
+ )
+
+ assert alert.interval == test_interval
+ assert type(alert.interval).__name__ == expected_name
+ else:
+ with pytest.raises(
+ ValueError,
+ match=f"Interval must be a `timedelta` or a
`VariableInterval`, received {expected_name}",
+ ):
+ DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=test_interval,
+ callback=TEST_DEADLINE_CALLBACK,
+ )
+
+ @pytest.mark.parametrize(
+ ("test_interval", "expected"),
+ [
+ pytest.param(3600, timedelta(seconds=3600), id="positive_int"),
+ pytest.param(-5, timedelta(seconds=-5), id="negative_int"),
+ pytest.param(0, timedelta(0), id="zero"),
+ pytest.param(0.1, timedelta(seconds=0.1), id="float"),
+ ],
+ )
+ def test_deadline_init_number_interval_is_deprecated_not_rejected(self,
test_interval, expected):
+ """A bare number was never a documented interval type, but it parses
today, so it is warned about."""
+ with pytest.warns(
+ RemovedInAirflow4Warning, match="Passing a number as a deadline
interval is deprecated"
+ ):
+ alert = DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=test_interval,
+ callback=TEST_DEADLINE_CALLBACK,
+ )
+
+ assert alert.interval == expected
+ assert type(alert.interval) is timedelta
+
+ def test_deadline_init_normalizes_timedelta_subclass(self):
+ """``pendulum.duration()`` has no serializer of its own, so it is
flattened to a ``timedelta``.
+
+ Without this it passes the isinstance check and then fails much later
in serde, which
+ dispatches on qualified class name and registers only
``datetime.timedelta``.
+ """
alert = DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
- interval=timedelta(hours=1),
- callback=callback_class(TEST_CALLBACK_PATH),
+ interval=pendulum.duration(hours=1),
+ callback=TEST_DEADLINE_CALLBACK,
)
- assert alert.callback is not None
- assert isinstance(alert.callback, callback_class)
- def test_deadline_alert_rejects_invalid_callback(self):
- """Test that DeadlineAlert rejects non-callback types."""
- with pytest.raises(ValueError, match="Callbacks of type str are not
currently supported"):
- DeadlineAlert(
- reference=DeadlineReference.DAGRUN_QUEUED_AT,
- interval=timedelta(hours=1),
- callback="not_a_callback", # type: ignore
- )
+ assert alert.interval == timedelta(hours=1)
+ assert type(alert.interval) is timedelta
class TestVariableInterval:
