potiuk commented on code in PR #73294:
URL: https://github.com/apache/airflow/pull/73294#discussion_r4066457568
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -154,14 +154,22 @@ def __init__(
callback: Callback,
name: str | None = None,
):
+ if not isinstance(interval, (timedelta, VariableInterval)):
Review Comment:
A bare number currently round-trips end to end, and this guard rejects it.
`_decode_deadline_interval` in
`airflow-core/src/airflow/serialization/decoders.py` has an explicit
backward-compat branch that turns an `int`/`float` into
`timedelta(seconds=...)`. On `main` today:
```python
DeadlineAlert(reference=..., interval=3600, callback=SyncCallback(...))
# serialized interval -> 3600
# deserialized interval -> datetime.timedelta(seconds=3600)
```
After this change that Dag stops parsing. `int` was never documented or
type-hinted, so tightening it is defensible — but it is a user-visible
behaviour change, and this PR carries `backport-to-v3-3-test`, which would land
it in a patch line.
Please make the call explicitly before merge, either:
- accept numbers and normalise them (`timedelta(seconds=interval)`), keeping
the authoring surface consistent with the decoder's back-compat branch; or
- keep the rejection, drop `backport-to-v3-3-test`, and add an
`airflow-core/newsfragments/` entry so the break is announced.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -154,14 +154,22 @@ def __init__(
callback: Callback,
name: str | None = None,
):
+ 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 (AsyncCallback, SyncCallback):
Review Comment:
This tuple is now the second copy of the permitted-callback set. The first
is `permitted` in `_decode_deadline_callback`
(`airflow-core/src/airflow/serialization/decoders.py`), which builds its
allow-list from `(AsyncCallback, SyncCallback)` too.
The comment above justifies this check precisely by mirroring that
allow-list, so the two drifting apart quietly invalidates the justification:
add a third callback type later and this guard rejects it at authoring time
even though the decoder would happily have accepted it — with no test anywhere
that would notice.
Please tie them to one source: export something like
`DEADLINE_CALLBACK_TYPES = (AsyncCallback, SyncCallback)` from
`airflow.sdk.definitions.callback` and build both sites from it. Core already
imports the SDK, so the dependency direction works.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -154,14 +154,22 @@ def __init__(
callback: Callback,
Review Comment:
Please narrow this annotation to `AsyncCallback | SyncCallback`.
After the check below, `Callback` is wider than what `__init__` actually
accepts: a `Callback` subclass type-checks cleanly and then raises at runtime.
The narrower annotation states the real contract, and lets mypy catch the
subclass case at static-check time instead of at Dag-parse time — which is the
same "fail earlier" goal this PR is pursuing.
##########
task-sdk/tests/task_sdk/definitions/test_deadline.py:
##########
@@ -142,30 +147,101 @@ def test_deadline_alert_in_set(self):
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):
- alert = DeadlineAlert(
- reference=DeadlineReference.DAGRUN_QUEUED_AT,
- interval=timedelta(hours=1),
- callback=callback_class(TEST_CALLBACK_PATH),
- )
- 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(
+ 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="not_a_callback", # type: ignore
+ 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_pass"),
+ [
+ pytest.param(timedelta(1), True, id="positive_timedelta_passes"),
+ pytest.param(timedelta(-1), True, id="negative_timedelta_passes"),
+ pytest.param(timedelta(0), True, id="zero_timedelta_passes"),
+ pytest.param(VariableInterval("var"), True,
id="VariableInterval_passes"),
+ pytest.param(1, False, id="int_fails"),
+ pytest.param(0.1, False, id="float_fails"),
+ pytest.param(True, False, id="bool_fails"),
+ pytest.param("str", False, id="string_fails"),
+ pytest.param(None, False, id="can_not_be_none"),
+ ],
+ )
+ def test_deadline_init_interval_type_checks(self, test_interval,
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) is type(test_interval)
+ else:
+ with pytest.raises(
+ ValueError, match="Interval must be a `timedelta` or a
`VariableInterval`, received"
Review Comment:
Please assert the reported type name here, the way
`test_deadline_init_callback_type_checks` does above.
As written, `match` stops at `received`, so the part of the message that
names the offending type is never exercised — the test would still pass if the
f-string reported the wrong type, or dropped it entirely. Given the whole point
of the new message is telling the Dag author *what* they passed, that's the
half worth pinning.
The params already carry what's needed; adding an `expected_name` column
alongside `expected_pass` mirrors the callback test exactly.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -154,14 +154,22 @@ def __init__(
callback: Callback,
name: str | None = None,
):
+ 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.
Review Comment:
This comment states the principle, and the interval check two lines above
doesn't follow it.
`isinstance(interval, timedelta)` accepts `timedelta` subclasses, and the
one Dag authors actually reach for is `pendulum.duration(...)` — Pendulum is
Airflow's own datetime library, and `pendulum.Duration` subclasses
`datetime.timedelta`. It passes this guard and then fails later:
```text
TypeError: Cannot serialize object of type <class
'pendulum.duration.Duration'>.
Give it a `serialize()` method and a `deserialize(data, version)`
staticmethod, ...
```
That is exactly the deferred, opaque failure this PR exists to move earlier
— so the interval guard doesn't yet do its job for the most likely wrong-ish
input.
Please fix by normalising rather than rejecting, which has the bonus of
making pendulum durations actually work:
```python
if isinstance(interval, timedelta) and type(interval) is not timedelta:
interval = timedelta(seconds=interval.total_seconds())
```
Rejecting instead is a fine answer too — but then it should be an exact-type
check matching the callback one below, so the two guards tell the same story.
--
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]
