hkc-8010 commented on code in PR #70370:
URL: https://github.com/apache/airflow/pull/70370#discussion_r4092156790
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ def decode_deadline_alert(encoded_data: dict):
)
+def decode_deadline_alert_model(deadline_alert: DeadlineAlertModel) ->
SerializedDeadlineAlert:
+ """
+ Decode a ``DeadlineAlert`` ORM row into its serialized representation.
+
+ :meta private:
+ """
+ return decode_deadline_alert(
+ {
+ DeadlineAlertFields.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
Review Comment:
Fixed, along with a test so it stays fixed.
You are right that nothing reads `.name` today, and that is exactly what
made this worth a test rather than just a one-line patch: with no caller and no
coverage, the next refactor drops it again and nothing goes red.
`TestDecodeDeadlineAlertModel` in `test_decoders.py` now asserts the round
trip. I checked it is a real differential by reverting the added line, which
fails with `-None +'my_deadline'`.
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ def decode_deadline_alert(encoded_data: dict):
)
+def decode_deadline_alert_model(deadline_alert: DeadlineAlertModel) ->
SerializedDeadlineAlert:
+ """
+ Decode a ``DeadlineAlert`` ORM row into its serialized representation.
+
+ :meta private:
+ """
+ return decode_deadline_alert(
+ {
+ DeadlineAlertFields.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+ }
+ )
+
+
+def resolve_deadline_alert_interval(
+ alert: SerializedDeadlineAlert, *, session: Session | None = None
Review Comment:
Done, it is now `*, session: Session` with no default.
Worth noting this costs nothing at the call sites: all three
(`definitions/dag.py`, `models/taskinstance.py`, and the tests) already passed
`session=` explicitly, so this is a pure tightening. Anyone who gets it wrong
in future now gets a `TypeError` at the call site instead of a silent rollback,
which is the point.
Separately, while you are here: the only thing still blocking this PR is
your `CHANGES_REQUESTED` from 2026-08-05 about the SDK coupling. That coupling
is gone since #71802 and #72651 landed. `models/taskinstance.py` no longer
imports from the SDK at all, and `known_sdk_imports_in_core.txt` is not in the
diff. Would you mind re-reviewing so it can move?
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ def decode_deadline_alert(encoded_data: dict):
)
+def decode_deadline_alert_model(deadline_alert: DeadlineAlertModel) ->
SerializedDeadlineAlert:
+ """
+ Decode a ``DeadlineAlert`` ORM row into its serialized representation.
+
+ :meta private:
+ """
+ return decode_deadline_alert(
+ {
+ DeadlineAlertFields.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+ }
+ )
+
+
+def resolve_deadline_alert_interval(
+ alert: SerializedDeadlineAlert, *, session: Session | None = None
+) -> datetime.timedelta:
+ """
+ Resolve a decoded alert's interval to a ``timedelta``.
+
+ A ``SerializedVariableInterval`` reads its Airflow Variable here, so this
is only called at
+ the point a deadline is actually calculated. It raises ``ValueError`` if
the Variable is
+ missing or is not an integer number of seconds.
+
+ :param alert: The decoded alert whose interval should be resolved.
+ :param session: Existing SQLAlchemy Session. Both callers run under the
scheduler's
+ ``prohibit_commit`` guard, so the open session has to reach
``Variable.get`` instead of
+ ``provide_session`` handing back the same scoped session and rolling
it back on exit.
Review Comment:
Agreed, that was four copies of the same paragraph drifting apart. The full
rationale now lives only in the `resolve_deadline_alert_interval()` docstring,
and the three tests point at it by name.
##########
airflow-core/tests/unit/models/test_taskinstance.py:
##########
@@ -4130,8 +4130,28 @@ async def empty_callback_for_deadline():
pass
-def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker,
session):
- """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT
deadlines."""
[email protected](
+ "use_variable_interval",
+ [
+ pytest.param(False, id="fixed_timedelta_interval"),
+ pytest.param(True, id="variable_interval"),
+ ],
+)
+def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker,
session, use_variable_interval):
+ """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT
deadlines.
+
+ Since Airflow 3.3 the ``deadline_alert.interval`` column is JSON (a
serialized ``timedelta``
+ or ``VariableInterval``), so the recalculation must decode it instead of
passing the raw value
+ to ``timedelta()``. Storing the interval via ``serialize`` here mirrors
production and covers
+ both interval kinds.
+ """
+ from airflow.models.variable import Variable
+ from airflow.sdk.definitions.deadline import VariableInterval
+ from airflow.sdk.serde import serialize
Review Comment:
Hoisted.
##########
airflow-core/tests/unit/models/test_taskinstance.py:
##########
@@ -4205,13 +4263,88 @@ def
test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
for deadline in deadlines_after:
if deadline.deadline_time !=
deadline_times_by_alert[deadline.deadline_alert_id]:
recalculated_count += 1
- deadline_alert = session.get(DeadlineAlertModel,
deadline.deadline_alert_id)
- expected_time = dag_run.queued_at +
datetime.timedelta(seconds=deadline_alert.interval)
+ expected_time = dag_run.queued_at +
expected_resolved_by_alert[deadline.deadline_alert_id]
assert deadline.deadline_time == expected_time
assert recalculated_count == 2
+def
test_clear_task_instances_skips_deadline_with_unresolvable_interval(dag_maker,
session):
+ """A variable-backed interval that cannot be resolved must not abort the
clear.
+
+ ``SerializedVariableInterval.resolve()`` raises ``ValueError`` when the
Airflow Variable is
+ missing or is not an integer, and that happens while the DAG run is being
cleared. The clear
+ should still go through, leaving the unresolvable deadline at its old time.
+ """
+ from airflow.models.variable import Variable
+ from airflow.sdk.definitions.deadline import VariableInterval
+ from airflow.sdk.serde import serialize
Review Comment:
Hoisted.
--
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]
