ferruzzi commented on code in PR #71802:
URL: https://github.com/apache/airflow/pull/71802#discussion_r3833081329
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1532,7 +1531,7 @@ def test_dagrun_deadline_variable_interval_stable(self,
_, mock_get, session, de
)
dag_run.dag = scheduler_dag
- # First update resolve interval to "5".
+ # First update resolves interval to "60".
Review Comment:
Nice catch
##########
airflow-core/tests/unit/serialization/test_serialized_objects.py:
##########
@@ -272,6 +277,48 @@ def equal_serialized_asset(a: SerializedAssetBase |
BaseAsset, b: SerializedAsse
return ensure_serialized_asset(a) == ensure_serialized_asset(b)
[email protected](
+ ("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):
Review Comment:
Maybe a nit, but maybe this test belongs in a new
`airflow-core/tests/unit/serialization/definitions/test_deadline.py` beside
`test_assets.py`, `test_param.py`, etc? Feel free to just resolve this if you
disagree, I can see an argument for either option.
##########
generated/known_sdk_imports_in_core.txt:
##########
@@ -25,8 +25,7 @@ airflow-core/src/airflow/providers_manager.py::5
airflow-core/src/airflow/secrets/__init__.py::1
airflow-core/src/airflow/serialization/decoders.py::2
airflow-core/src/airflow/serialization/definitions/baseoperator.py::1
-airflow-core/src/airflow/serialization/definitions/dag.py::2
-airflow-core/src/airflow/serialization/definitions/deadline.py::1
Review Comment:
Nice bonus!
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -203,16 +204,19 @@ def decode_deadline_alert(encoded_data: dict):
"from a version that supports VariableInterval. Downgrade is not
fully reversible."
)
- interval: datetime.timedelta | VariableInterval
+ interval: datetime.timedelta | SerializedVariableInterval
# Backward compatibility: previously interval was stored as
total_seconds() (float/int).
# Handle numeric values by converting to timedelta.
if isinstance(raw_interval, (int, float)):
interval = datetime.timedelta(seconds=raw_interval)
else:
deserialized = deserialize(raw_interval)
- if isinstance(deserialized, (datetime.timedelta, VariableInterval)):
+
+ if isinstance(deserialized, datetime.timedelta):
interval = deserialized
+ elif isinstance(deserialized, VariableInterval):
+ interval = SerializedVariableInterval(key=deserialized.key)
else:
Review Comment:
Doesn't this need a branch (and a new test) for the new
`SerializedVariableInterval` since the encoder accepts it?
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -436,21 +434,3 @@ class VariableInterval:
"""
key: str
-
- def resolve(self) -> timedelta:
Review Comment:
Pretty sure this is released in 3.3.0 so it will need to be deprecated,
without the `seconds <= 0` check.
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -380,11 +378,35 @@ def _fetch_from_db(column, *, session: Session, dag_id:
str, run_id: str) -> dat
return result
[email protected](frozen=True)
+class SerializedVariableInterval:
+ """Core-side serialized representation of a variable-backed deadline
interval."""
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ from airflow.models.variable import Variable
+
+ try:
+ value = Variable.get(self.key)
+ except KeyError as e:
Review Comment:
I only know this because I was just working in that code path, but
Variable.get() can also raise `AirflowRuntimeError` on some paths, you likely
want to catch that here as well? But I guess that would mean importing from
sdk into core, so maybe not....
##########
airflow-core/newsfragments/71802.improvement.rst:
##########
@@ -0,0 +1 @@
+Add a core-side ``SerializedVariableInterval`` representation for deadline
alerts, with ``VariableInterval`` converted to its serialized counterpart
during deserialization and resolved in core during deadline evaluation;
variable-backed intervals now also support zero and negative offsets to align
with existing ``timedelta`` semantics.
Review Comment:
I'd suggest filing this as a bugfix rather than an improvement and perhaps
rephrasing this a little; Intervals were always expected to allow a negative
value, requiring a positive value for VariableInterval was a breaking change
that went against the existing pattern.
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -380,11 +378,35 @@ def _fetch_from_db(column, *, session: Session, dag_id:
str, run_id: str) -> dat
return result
[email protected](frozen=True)
+class SerializedVariableInterval:
+ """Core-side serialized representation of a variable-backed deadline
interval."""
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ from airflow.models.variable import Variable
Review Comment:
Does this need to be a local import?
--
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]
