Keep VariableInterval.resolve() as a deprecated compatibility shim instead of removing the public method outright. Move SerializedVariableInterval resolution tests into a dedicated deadline serialization test module and add compatibility coverage for the deprecated VariableInterval.resolve() path.

Author: Sameer Mesiah

airflow-core/newsfragments/71802.bugfix.rst

@@ -0,0 +1 @@
+Fixed ``VariableInterval`` deadline intervals to allow zero and negative offsets, matching the existing ``timedelta`` interval semantics. ``VariableInterval`` is now converted to the core-side ``SerializedVariableInterval`` representation during deadline deserialization and resolved during deadline evaluation.

airflow-core/newsfragments/71802.improvement.rst

@@ -215,6 +215,8 @@ def decode_deadline_alert(encoded_data: dict):
 
         if isinstance(deserialized, datetime.timedelta):
             interval = deserialized
+        elif isinstance(deserialized, SerializedVariableInterval):
+            interval = deserialized
         elif isinstance(deserialized, VariableInterval):
             interval = SerializedVariableInterval(key=deserialized.key)
         else:

airflow-core/src/airflow/serialization/decoders.py

@@ -28,6 +28,7 @@
 
 from airflow._shared.timezones import timezone
 from airflow.models.deadline import classproperty
+from airflow.models.variable import Variable
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import provide_session
 from airflow.utils.sqlalchemy import get_dialect_name
@@ -385,7 +386,6 @@ class SerializedVariableInterval:
     key: str
 
     def resolve(self) -> timedelta:
-        from airflow.models.variable import Variable
 
         try:
             value = Variable.get(self.key)

airflow-core/src/airflow/serialization/definitions/deadline.py

@@ -0,0 +1,67 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+import pytest
+
+from airflow.models.variable import Variable
+from airflow.serialization.definitions.deadline import SerializedVariableInterval
+
+
+class TestVariableInterval:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            ("3", timedelta(seconds=3)),
+            ("10", timedelta(seconds=10)),
+            ("05", timedelta(seconds=5)),
+            ("0", timedelta(0)),
+            ("-5", timedelta(seconds=-5)),
+        ],
+    )
+    def test_resolve_valid(self, mocker, value, expected):
+        mocker.patch.object(Variable, "get", return_value=value)
+
+        interval = SerializedVariableInterval(key="test_interval")
+
+        assert interval.resolve() == expected
+
+    @pytest.mark.parametrize(
+        ("value", "raise_missing", "match"),
+        [
+            (None, True, "not found"),
+            ("abc", False, "must be an integer"),
+            ("", False, "must be an integer"),
+        ],
+    )
+    def test_resolve_invalid(self, mocker, value, raise_missing, match):
+        if raise_missing:
+            mocker.patch.object(
+                Variable,
+                "get",
+                side_effect=KeyError("test_interval"),
+            )
+        else:
+            mocker.patch.object(Variable, "get", return_value=value)
+
+        interval = SerializedVariableInterval(key="test_interval")
+
+        with pytest.raises(ValueError, match=match):
+            interval.resolve()

airflow-core/tests/unit/serialization/definitions/test_deadline.py

@@ -43,7 +43,6 @@
 )
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG
-from airflow.models.variable import Variable
 from airflow.models.xcom_arg import XComArg
 from airflow.partition_mappers.identity import IdentityMapper as CoreIdentityMapper
 from airflow.partition_mappers.temporal import (
@@ -80,6 +79,7 @@
     AsyncCallback,
     DeadlineAlert,
     DeadlineReference,
+    VariableInterval,
 )
 from airflow.sdk.definitions.decorators import task
 from airflow.sdk.definitions.operator_resources import Resources
@@ -277,48 +277,6 @@ def equal_serialized_asset(a: SerializedAssetBase | BaseAsset, b: SerializedAsse
     return ensure_serialized_asset(a) == ensure_serialized_asset(b)
 
 
-@pytest.mark.parametrize(
-    ("value", "expected"),
-    [
-        ("3", timedelta(seconds=3)),
-        ("10", timedelta(seconds=10)),
-        ("05", timedelta(seconds=5)),
-        ("0", timedelta(0)),
-        ("-5", timedelta(seconds=-5)),
-    ],
-)
-def test_serialized_variable_interval_resolve_valid(mocker, value, expected):
-    mocker.patch.object(Variable, "get", return_value=value)
-
-    interval = SerializedVariableInterval(key="test_interval")
-
-    assert interval.resolve() == expected
-
-
-@pytest.mark.parametrize(
-    ("value", "raise_missing", "match"),
-    [
-        (None, True, "not found"),
-        ("abc", False, "must be an integer"),
-        ("", False, "must be an integer"),
-    ],
-)
-def test_serialized_variable_interval_resolve_invalid(mocker, value, raise_missing, match):
-    if raise_missing:
-        mocker.patch.object(
-            Variable,
-            "get",
-            side_effect=KeyError("test_interval"),
-        )
-    else:
-        mocker.patch.object(Variable, "get", return_value=value)
-
-    interval = SerializedVariableInterval(key="test_interval")
-
-    with pytest.raises(ValueError, match=match):
-        interval.resolve()
-
-
 def equal_serialized_deadline_alert(
     a: DeadlineAlert | SerializedDeadlineAlert,
     b: DeadlineAlert | SerializedDeadlineAlert,
@@ -539,13 +497,33 @@ def test_serialize_deserialize_connection():
 
 
 @pytest.mark.parametrize("reference", REFERENCE_TYPES)
-def test_serialize_deserialize_deadline_alert(reference):
+@pytest.mark.parametrize(
+    ("interval", "expected_interval"),
+    [
+        pytest.param(
+            timedelta(hours=1),
+            timedelta(hours=1),
+            id="timedelta",
+        ),
+        pytest.param(
+            VariableInterval("deadline_seconds"),
+            SerializedVariableInterval("deadline_seconds"),
+            id="sdk_variable_interval",
+        ),
+        pytest.param(
+            SerializedVariableInterval("deadline_seconds"),
+            SerializedVariableInterval("deadline_seconds"),
+            id="serialized_variable_interval",
+        ),
+    ],
+)
+def test_serialize_deserialize_deadline_alert(reference, interval, expected_interval):
     public_deadline_alert_fields = {
         field.lower() for field in vars(DeadlineAlertFields) if not field.startswith("_")
     }
     original = DeadlineAlert(
         reference=reference,
-        interval=timedelta(hours=1),
+        interval=interval,
         callback=AsyncCallback(empty_callback_for_deadline, kwargs=TEST_CALLBACK_KWARGS),
     )
 
@@ -556,7 +534,7 @@ def test_serialize_deserialize_deadline_alert(reference):
 
     deserialized = BaseSerialization.deserialize(serialized)
     assert deserialized.reference.serialize_reference() == reference.serialize_reference()
-    assert deserialized.interval == original.interval
+    assert deserialized.interval == expected_interval
     assert deserialized.callback == original.callback
 
 

airflow-core/tests/unit/serialization/test_serialized_objects.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import logging
+import warnings
 from abc import ABC
 from dataclasses import dataclass
 from datetime import datetime, timedelta
@@ -25,6 +26,8 @@
 import attrs
 
 from airflow.sdk.definitions.callback import AsyncCallback, Callback, SyncCallback
+from airflow.sdk.definitions.variable import Variable
+from airflow.sdk.exceptions import AirflowRuntimeError
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -434,3 +437,25 @@ class VariableInterval:
     """
 
     key: str
+
+    def resolve(self) -> timedelta:
+        warnings.warn(
+            "VariableInterval.resolve() is deprecated and will be removed in a future release. "
+            "Deadline interval resolution is handled internally during deadline evaluation.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
+        try:
+            value = Variable.get(self.key)
+        except AirflowRuntimeError as e:
+            raise ValueError(f"VariableInterval '{self.key}' not found") from e
+
+        try:
+            seconds = int(value)
+        except (TypeError, ValueError) as e:
+            raise ValueError(
+                f"VariableInterval '{self.key}' must be an integer (seconds), got: {value!r}"
+            ) from e
+
+        return timedelta(seconds=seconds)

task-sdk/src/airflow/sdk/definitions/deadline.py

@@ -17,12 +17,15 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta
+from unittest import mock
 
 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
+from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference, VariableInterval
+from airflow.sdk.definitions.variable import Variable
+from airflow.sdk.exceptions import AirflowRuntimeError
 
 DAG_ID = "dag_id_1"
 RUN_ID = 1
@@ -162,3 +165,51 @@ def test_deadline_alert_rejects_invalid_callback(self):
                 interval=timedelta(hours=1),
                 callback="not_a_callback",  # type: ignore
             )
+
+
+class TestVariableInterval:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            ("3", timedelta(seconds=3)),
+            ("10", timedelta(seconds=10)),
+            ("05", timedelta(seconds=5)),
+            ("0", timedelta(0)),
+            ("-5", timedelta(seconds=-5)),
+        ],
+    )
+    def test_resolve_valid(self, mocker, value, expected):
+        mocker.patch.object(Variable, "get", return_value=value)
+
+        interval = VariableInterval(key="test_interval")
+
+        with pytest.warns(DeprecationWarning, match="VariableInterval.resolve"):
+            assert interval.resolve() == expected
+
+    @pytest.mark.parametrize(
+        ("value", "raise_runtime", "match"),
+        [
+            (None, True, "not found"),
+            ("abc", False, "must be an integer"),
+            ("", False, "must be an integer"),
+        ],
+    )
+    def test_resolve_invalid(self, mocker, value, raise_runtime, match):
+        if raise_runtime:
+            mock_err = mock.Mock()
+            mock_err.error.value = "MISSING"
+            mock_err.detail = "missing"
+
+            mocker.patch.object(
+                Variable,
+                "get",
+                side_effect=AirflowRuntimeError(mock_err),
+            )
+        else:
+            mocker.patch.object(Variable, "get", return_value=value)
+
+        interval = VariableInterval(key="test_interval")
+
+        with pytest.warns(DeprecationWarning, match="VariableInterval.resolve"):
+            with pytest.raises(ValueError, match=match):
+                interval.resolve()