You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "SameerMesiah97 (via GitHub)" <gi...@apache.org> on 2026/04/06 00:21:02 UTC
[PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
SameerMesiah97 opened a new pull request, #64751:
URL: https://github.com/apache/airflow/pull/64751
**Description**
This change adds support for **dynamic interval resolution** in `encode_deadline_alert` by allowing interval objects that implement a `.resolve()` method (e.g. `VariableInterval`) to be evaluated during serialization. The encoder now detects and invokes `.resolve()` when present, converting the result into a `timedelta` for serialization while preserving existing behavior for `timedelta` intervals.
The encoder is also made **idempotent**, returning early when given an already-serialized dict to prevent double-encoding during DAG re-serialization. Additionally, a `VariableInterval` class is introduced under `airflow.sdk.definitions.deadline`, enabling intervals to be sourced from Airflow Variables (interpreted as minutes).
**Rationale**
`DeadlineAlert `intervals are part of DAG configuration and are resolved at **DAG parse / serialization time**, not at runtime. Supporting runtime resolution would require introducing a separate execution-time mechanism for interval evaluation, diverging from how `timedelta` intervals are handled today and increasing system complexity for a relatively small, opt-in feature.
This change instead resolves intervals at **serialization time**, which is the last point before the value is persisted to the database. This ensures that both static (`timedelta`) and dynamic (`VariableInterval`) intervals are materialized in a single, consistent phase without introducing any changes to scheduler or runtime behavior. As a result, interval values remain stable for the lifetime of a DagRun while keeping the implementation minimal and aligned with existing architecture.
**Tests**
Added unit tests verifying that:
* `timedelta` intervals are correctly serialized.
* `.resolve()` is invoked for dynamic interval objects.
* encoding is idempotent (`encode(x) == encode(encode(x))`).
* callback fields are preserved during encoding.
* `VariableInterval.resolve()` correctly converts a Variable value (in minutes) to a `timedelta` (mocked via `Variable.get`).
**Documentation**
* Modified 'how-to' docs to mention that `DeadlineAlert `intervals may be dynamically resolved via VariableInterval.
* Added docstring for `VariableInterval` with usage example and behavior notes.
**Backwards Compatibility**
This change is fully backwards compatible. Existing `timedelta`-based intervals continue to behave as before, and dynamic interval resolution is strictly opt-in via objects implementing `.resolve()` (e.g. `VariableInterval`).
Closes: #63852
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3114494538
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
Review Comment:
Yes. I have run multiple tests with `VariableInterval` and `timedelta` during development and can confirm the deadline interval is being set via Variables when `VariableInterval` is passsed. I can do another test run and post the results if you want.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086234645
##########
airflow-core/src/airflow/migrations/versions/0112_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ );
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}';
+ """
+ )
+ return
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ postgresql_using="to_json(interval)",
+ existing_nullable=False,
+ )
+ else:
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ existing_nullable=False,
+ )
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number'
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ )
+ """)
+
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}'
+ """)
+
+
+def downgrade():
+ """Revert deadline interval back to float."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual downgrade required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END;
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END;
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END;
+ """
+ )
+ return
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END
+ """)
Review Comment:
Now, only seriallized `timedelta` objects will be converted to interval values. Please see below:
```
UPDATE deadline_alert
SET interval =
CASE
WHEN jsonb_typeof(interval::jsonb) = 'number'
THEN interval
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
THEN to_json((interval->>'__data__')::double precision)
ELSE NULL
END
```
Serialized `VariableInterval` objects will become NULL in the downgrade path. This will break `DeadlineAlerts` with interval set to `VariableInterval`, which is a deliberate tradeoff in this scenario. A guard has been added at the deserialization layer to handle situations where the metadata DB has been downgraded when serialized `VariableInterval` objects exist, resulting in the interval being deserialized as `None`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3134355697
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
Sorry, one last thing I just noticed. I am pretty sure I ran intot his before and `int(timedelta)` will raise TypeError? If it does, then you should be able to use `int(self.interval.total_seconds())` here and elsewhere instead now that it's seconds and not minutes.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3817272921
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
+
+ elif isinstance(self.interval, datetime.timedelta):
+ interval_seconds = int(self.interval.total_seconds())
+
+ if interval_seconds is None:
+ interval_display = "dynamic"
+ elif interval_seconds >= 3600:
Review Comment:
Cool, looks like that fix landed on Monday. Thanks for replying here.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3761122255
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
+
+ elif isinstance(self.interval, datetime.timedelta):
+ interval_seconds = int(self.interval.total_seconds())
+
+ if interval_seconds is None:
+ interval_display = "dynamic"
+ elif interval_seconds >= 3600:
Review Comment:
This is a regression, changing `interval` from `float` to `dict/JSON` means that not only can the first isinstance not possibly catch, it will raise an exception since you meant to use `datetime.timedelta` but instead you are using `datetime.datetime.timedelta`
`test_deadline_alert_repr` misses it because `DEADLINE_INTERVAL = 60` short-circuits on the first branch.
untested claude cuggestion:
```python
def __repr__(self):
from airflow.sdk.definitions.deadline import VariableInterval
from airflow.sdk.serde import deserialize
interval = self.interval
if isinstance(interval, dict):
try:
interval = deserialize(interval)
except Exception: # noqa: BLE001 - a repr must never raise
interval = None
interval_seconds = None
interval_display = "unknown"
if isinstance(interval, VariableInterval):
interval_display = f"var:{interval.key}"
elif isinstance(interval, timedelta):
interval_seconds = int(interval.total_seconds())
elif isinstance(interval, (int, float)):
# Bare seconds predate 3.3.0; migration 0117 should have converted these.
interval_seconds = int(interval)
if interval_seconds is not None:
if abs(interval_seconds) >= 3600:
interval_display = f"{interval_seconds // 3600}h"
elif abs(interval_seconds) >= 60:
interval_display = f"{interval_seconds // 60}m"
else:
interval_display = f"{interval_seconds}s"
return (
f"[DeadlineAlert] "
f"id={str(self.id)[:8]}, "
f"created_at={self.created_at}, "
f"name={self.name or 'Unnamed'}, "
f"reference={self.reference}, "
f"interval={interval_display}, "
f"callback={self.callback_def}"
)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3760773539
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -342,3 +346,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as seconds and
+ converted into a ``timedelta`` object.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_seconds"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ 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
+
+ if seconds <= 0:
+ raise ValueError(f"VariableInterval '{self.key}' must be > 0, got: {seconds}")
Review Comment:
I just noticed this, months later. This is a definite bug. Interval can be negative and the docs advertise this as an expected use-case. For example "30 minutes before the next scheduled run" or "20 minutes before the morning meeting" are both valid uses for a negative interval.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086241296
##########
airflow-core/src/airflow/migrations/versions/0112_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
Review Comment:
Offline mode instructions have been updated so that they mirror the steps for online mode.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221041961
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
I apologise. It appears I overlooked the first int(self.interval).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3114334243
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
Review Comment:
Have you tested this in an actual DAG Processor context? The unit tests all mock Variable.get so it would be good to confirm with manual verification.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
Review Comment:
Should this be minutes? I could be wrong, but I feel like we usually measure everything in seconds in Airflow.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3721328062
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -41,6 +41,7 @@
from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.models.tasklog import LogTemplate
+from airflow.sdk.definitions.deadline import VariableInterval
Review Comment:
This isnt right.
This breaks the client server separation pattern that this same file already follows for reference: `SerializedReferenceModels.SerializedBaseDeadlineReference` gives core a counterpart for each SDK reference type so core code never needs to import or isinstance check against the SDK classes directly.
`VariableInterval` does not have that counterpart, so core ends up depending on task SDK internals at runtime for this one field. We should be adding a `SerializedVariableInterval` (or similar) alongside `SerializedReferenceModels`, with `resolve()` living core side, so `decode_deadline_alert` returns a core type instead of the raw sdk dataclass.
Flagging now because #70370 just had to reproduce the same SDK import to fix a separate bug, so the gap is starting to spread.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086165218
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +347,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define
Review Comment:
I have set `@attrs.define(frozen=True)` for `VariableInterval` which should ensure sufficient hashability.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3114518059
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
Review Comment:
> Should this be minutes? I could be wrong, but I feel like we usually measure everything in seconds in Airflow.
I have no strong opinions either way but I chose minutes because the publicly documented example DAGs for `DeadlineAlerts` use minutes for timedelta. Please see the below:
```
from datetime import datetime, timedelta
from airflow.sdk import AsyncCallback, DAG, DeadlineAlert, DeadlineReference
from airflow.providers.slack.notifications.slack_webhook import SlackWebhookNotifier
from airflow.providers.standard.operators.empty import EmptyOperator
with DAG(
dag_id="deadline_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(minutes=15),
callback=AsyncCallback(
SlackWebhookNotifier,
kwargs={
"text": "π¨ Dag {{ dag_run.dag_id }} missed deadline at {{ deadline.deadline_time }}. DagRun: {{ dag_run }}"
},
),
),
):
EmptyOperator(task_id="example_task")
```
Here is the link to the documentation:
https://airflow.apache.org/docs/apache-airflow/stable/howto/deadline-alerts.html
Of course, the downside is that we will be losing precision but my understanding is that most users will not set SLAs at the second level as they will be running batches that will run for several minutes at the very least.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221041961
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
I apologise. This bug must have been re-introduced after my last force push. I will fix it again.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #64751:
URL: https://github.com/apache/airflow/pull/64751
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086240155
##########
airflow-core/newsfragments/64751.feature.rst:
##########
@@ -0,0 +1 @@
+Allow DeadlineAlert intervals to be dynamically resolved at DAG parse time using objects such as VariableInterval.
Review Comment:
This has been changed to reflect the actual implementation now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3119692864
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
Review Comment:
Just checking. I'll take your word for it. :smiley:
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086291528
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,11 +50,19 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
+
+ elif isinstance(self.interval, datetime.timedelta):
+ interval_seconds = int(self.interval.total_seconds())
+
+ else:
+ interval_display = "dynamic"
if interval_seconds >= 3600:
interval_display = f"{interval_seconds // 3600}h"
Review Comment:
`interval_seconds` is now set to `None` first and branched further down.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3825849451
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -41,6 +41,7 @@
from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.models.tasklog import LogTemplate
+from airflow.sdk.definitions.deadline import VariableInterval
Review Comment:
@amoghrajesh
I have opened a follow-up PR #71802 to address this.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3808102587
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -342,3 +346,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as seconds and
+ converted into a ``timedelta`` object.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_seconds"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ 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
+
+ if seconds <= 0:
+ raise ValueError(f"VariableInterval '{self.key}' must be > 0, got: {seconds}")
Review Comment:
I have opened a follow-up PR that fixes this as well. Please see PR #71802.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4195160708
@seanghaeli may want to have an eye on this one as well.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3066478875
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,11 +50,19 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
+
+ elif isinstance(self.interval, datetime.timedelta):
+ interval_seconds = int(self.interval.total_seconds())
+
+ else:
+ interval_display = "dynamic"
if interval_seconds >= 3600:
interval_display = f"{interval_seconds // 3600}h"
Review Comment:
`interval_seconds` is not set in the final `else` branch, but is used unconditionally afterward (`if interval_seconds >= 3600`). For non-numeric / non-`timedelta` intervals (e.g., serialized dicts or dynamic interval objects), `__repr__` will raise `UnboundLocalError`. Restructure `__repr__` to either return/format immediately for dynamic intervals or initialize `interval_seconds` to a sentinel and branch safely.
##########
airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ );
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}';
+ """
+ )
+ return
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ postgresql_using="to_json(interval)",
+ existing_nullable=False,
+ )
+ else:
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ existing_nullable=False,
+ )
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number'
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ )
+ """)
+
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}'
+ """)
+
+
+def downgrade():
+ """Revert deadline interval back to float."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual downgrade required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END;
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END;
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END;
+ """
+ )
+ return
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END
+ """)
+
+ # Serialized VariableInterval objects do not contain a numeric "__data__" field
+ # and therefore cannot be converted back to a float representation.
+ # During downgrade, only timedelta-style serialized values are converted.
+ # Other serialized interval types (e.g. VariableInterval) are not explicitly
+ # handled here and will fall through to the generic casting branch, which may
+ # result in NULL or backend-specific casting behavior.
+ else:
+ # Detect availability of SQLite JSON functions (JSON1 extension).
+ # If available, use json_extract for robust parsing.
+ # Otherwise, fall back to string-based extraction.
+ json_functions_available = False
+ try:
+ conn.execute(sa.text("SELECT JSON_SET('{}', '$.test', 'value')")).fetchone()
+ json_functions_available = True
+ print("SQLite JSON functions detected, using optimized SQL approach")
+ except Exception:
+ print("SQLite JSON functions not available, using Python fallback for JSON processing")
+
+ if json_functions_available:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END
+ """)
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN instr(interval, '__data__') > 0
+ THEN CAST(
+ substr(
+ interval,
+ instr(interval, '__data__') +
+ instr(substr(interval, instr(interval, '__data__')), ':')
+ ) AS FLOAT
+ )
+ ELSE CAST(interval AS FLOAT)
+ END
+ """)
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.JSON(),
+ type_=sa.FLOAT(),
+ postgresql_using="(interval->>'__data__')::double precision",
+ existing_nullable=False,
+ )
Review Comment:
The PostgreSQL downgrade path will fail for serialized non-`timedelta` JSON objects (e.g., `VariableInterval`) because the `ELSE to_json((interval::text)::double precision)` branch attempts to cast a JSON objectβs text (like `{\"__classname__\": ...}`) to `double precision`, which will error and abort the migration. Similarly, `postgresql_using=\"(interval->>'__data__')::double precision\"` will fail when `__data__` is absent. Update the downgrade to only convert rows that are known numeric or known `datetime.timedelta` serialized shapes (e.g., check `interval->>'__classname__' = 'datetime.timedelta'` or `jsonb_typeof(interval::jsonb) = 'number'`) and decide an explicit fallback for unsupported dynamic objects (e.g., set to NULL / a safe default / raise with a targeted message).
##########
airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ );
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}';
+ """
+ )
+ return
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ postgresql_using="to_json(interval)",
+ existing_nullable=False,
+ )
+ else:
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ existing_nullable=False,
+ )
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number'
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ )
+ """)
+
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}'
+ """)
+
+
+def downgrade():
+ """Revert deadline interval back to float."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual downgrade required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END;
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END;
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END;
+ """
+ )
+ return
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END
+ """)
Review Comment:
The PostgreSQL downgrade path will fail for serialized non-`timedelta` JSON objects (e.g., `VariableInterval`) because the `ELSE to_json((interval::text)::double precision)` branch attempts to cast a JSON objectβs text (like `{\"__classname__\": ...}`) to `double precision`, which will error and abort the migration. Similarly, `postgresql_using=\"(interval->>'__data__')::double precision\"` will fail when `__data__` is absent. Update the downgrade to only convert rows that are known numeric or known `datetime.timedelta` serialized shapes (e.g., check `interval->>'__classname__' = 'datetime.timedelta'` or `jsonb_typeof(interval::jsonb) = 'number'`) and decide an explicit fallback for unsupported dynamic objects (e.g., set to NULL / a safe default / raise with a targeted message).
##########
airflow-core/newsfragments/64751.feature.rst:
##########
@@ -0,0 +1 @@
+Allow DeadlineAlert intervals to be dynamically resolved at DAG parse time using objects such as VariableInterval.
Review Comment:
The newsfragment says intervals are resolved at 'DAG parse time', but the PR description and implementation resolve `VariableInterval` during DagRun deadline evaluation/creation. Please update the fragment text to match the actual behavior to avoid user confusion.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +347,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
Review Comment:
`DeadlineAlert.__hash__` hashes `self.interval`, but `@attrs.define` defaults can make instances unhashable when `eq=True` (common default), which will raise `TypeError: unhashable type: 'VariableInterval'` when a `DeadlineAlert` is hashed. Make `VariableInterval` explicitly hashable (e.g., `@attrs.define(frozen=True)` or an explicit `__hash__`) or update `DeadlineAlert.__hash__` to avoid hashing non-hashable intervals.
```suggestion
def __hash__(self) -> int:
return hash(self.key)
```
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -162,9 +162,18 @@ def decode_deadline_alert(encoded_data: dict):
reference_data = data[DeadlineAlertFields.REFERENCE]
reference = decode_deadline_reference(reference_data)
+ raw_interval = data[DeadlineAlertFields.INTERVAL]
+
+ # 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:
+ interval = cast("datetime.timedelta", deserialize(raw_interval))
Review Comment:
The `cast(\"datetime.timedelta\", ...)` is misleading now that `deserialize(raw_interval)` can return non-`timedelta` interval objects (e.g., `VariableInterval`). This doesnβt affect runtime behavior, but it hides type mismatches from linters/type-checkers and makes it easier to accidentally treat the result as always a `timedelta`. Prefer widening the type (e.g., `timedelta | VariableInterval | ...`) and removing or adjusting the cast accordingly.
##########
airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
Review Comment:
The offline-mode 'Manual conversion required' instructions appear inconsistent with the actual upgrade sequence: they do not include the required `ALTER TABLE`/type change step, and the PostgreSQL `WHERE jsonb_typeof(interval::jsonb)` predicate implies `interval` is already JSON (but offline mode returns before any conversion). Consider updating the offline instructions to be a complete, executable sequence (type change + transformation) and ensure predicates match the pre/post column type at each step.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +347,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define
Review Comment:
`DeadlineAlert.__hash__` hashes `self.interval`, but `@attrs.define` defaults can make instances unhashable when `eq=True` (common default), which will raise `TypeError: unhashable type: 'VariableInterval'` when a `DeadlineAlert` is hashed. Make `VariableInterval` explicitly hashable (e.g., `@attrs.define(frozen=True)` or an explicit `__hash__`) or update `DeadlineAlert.__hash__` to avoid hashing non-hashable intervals.
```suggestion
@attrs.define(unsafe_hash=True)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3114762335
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ try:
+ value = Variable.get(self.key)
+ except KeyError as e:
+ raise ValueError(f"VariableInterval '{self.key}' not found") from e
+
Review Comment:
`Variable.get()` raises `AirflowRuntimeError` when a key is missing, not `KeyError` (see `variable.py:47-58`). Since no `default` is passed here, the `AirflowRuntimeError` re-raises and this `except KeyError` block never triggers. The test passes because it mocks `Variable.get` with `side_effect=KeyError("missing")`.
Should be:
```python
from airflow.sdk.exceptions import AirflowRuntimeError
try:
value = Variable.get(self.key)
except AirflowRuntimeError as e:
raise ValueError(f"VariableInterval '{self.key}\ not found") from e
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4284864538
Ready for review.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3142357936
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
Fixed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3220883140
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
Is it? I still see `self.interval` here, have you confirmed that this works as written?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221041961
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
I apologise. It appears I overlooked the first int(self.interval).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221220227
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
> Is it? I still see `self.interval` here, have you confirmed that this works as written?
I believe this will only apply to line 64 below as the condition will check that `self.interval` is a `timedelta` object before applying `total_seconds`. Line 60-61 is for floats and integers that have not been converted to timedelta objects.
I have tested the following DAG shape with both interval passed using `timedelta` and the new `VariableInterval` object and can verify that it is working as expected:
```
def long_running_task ():
print ("Starting long-running task")
time.sleep (120)
print ("Ending long-running task")
with DAG(
dag_id="deadline_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval= [VariableInterval (key = 'deadline_seconds') / timedelta (seconds = 60],
# deadline_seconds is set to 60.
callback=AsyncCallback(
custom_alert_callback
),
),
):
PythonOperator(task_id="long_running_task",
python_callable= long_running_task)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4195213673
> @seanghaeli may want to have an eye on this one as well.
This is currently not reviewable. Please give me a bit more time to smoothen out a few rough edges.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4190477790
This is in draft but a few thoughts on this:
1. Side effects in the encoder break the serialization contract
Every other encoder in encoders.py is a pure objectβdict transformation β no network calls, no database access. This PR makes encode_deadline_alert the only one that reaches out to external systems (Variable.get())
This means serialization is no longer deterministic β the same DAG object produces different output depending on when you serialize it. It also means serialization can fail for reasons unrelated to encoding (Variable doesn't exist, API server timeout), surfacing as a confusing DAG parse failure.
2. If a user changes the Variable in the UI, nothing happens until the DAG Processor re-parses the file on its next cycle. The re-parse is timer-based (min_file_process_interval), not triggered by the Variable change β so the delay is unpredictable and there's no feedback that it took effect.
@ferruzzi has recommended in the original issue to handle this when the deadline is calculated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4299430260
@SameerMesiah97 This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Failing CI**: `Task SDK tests / task-sdk:P3.10 tests` is failing. Reproduce locally with `breeze testing task-sdk-tests` and investigate.
- :x: **Unresolved review comments** (1 thread): please walk through each unresolved review thread. **Even if a suggestion looks incorrect or irrelevant** β and some of them will be, especially any comments left by automated reviewers like GitHub Copilot β it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. Leaving threads unaddressed for weeks blocks the PR from moving forward.
**What to do next:**
- Walk through each unresolved review thread and respond as described above.
- Fix the Task SDK test failure and push.
- Make sure static checks pass locally: `prek run --from-ref main --stage pre-commit`.
- Mark the PR as "Ready for review" when you're done.
Converting a PR to draft is **not** a rejection β it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush β take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4300588872
> Quick follow-up to the triage comment above β one clarification on the "Unresolved review comments" item:
>
> Once you believe a thread has been addressed β whether by pushing a fix, or by replying in-thread with an explanation of why the suggestion doesn't apply β **please mark the thread as resolved yourself** by clicking the "Resolve conversation" button at the bottom of each thread. Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as "still waiting on the author" and keeps the PR from moving forward. The author doing the resolve-click is the expected convention on this project.
Understood.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3127469181
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
Review Comment:
Seconds are being used now.
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ try:
+ value = Variable.get(self.key)
+ except KeyError as e:
+ raise ValueError(f"VariableInterval '{self.key}' not found") from e
+
Review Comment:
Fixed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086241296
##########
airflow-core/src/airflow/migrations/versions/0112_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
Review Comment:
Instructions have been updated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3086269863
##########
airflow-core/src/airflow/migrations/versions/0112_3_3_0_change_deadline_interval_to_json.py:
##########
@@ -0,0 +1,248 @@
+#
+# 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.
+
+"""
+Change type of interval in Deadline Alerts table to JSON.
+
+Revision ID: 82f208dbbad5
+Revises: a4c2d171ae18
+Create Date: 2026-04-06 16:55:46.517409
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+# revision identifiers, used by Alembic.
+revision = "82f208dbbad5"
+down_revision = "a4c2d171ae18"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+ """Apply change deadline interval to JSON."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual conversion required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number';
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ );
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}';
+ """
+ )
+ return
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ postgresql_using="to_json(interval)",
+ existing_nullable=False,
+ )
+ else:
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.FLOAT(),
+ type_=sa.JSON(),
+ existing_nullable=False,
+ )
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval = json_build_object(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', (interval::text)::float
+ )
+ WHERE jsonb_typeof(interval::jsonb) = 'number'
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` = JSON_OBJECT(
+ '__classname__', 'datetime.timedelta',
+ '__version__', 2,
+ '__data__', `interval`
+ )
+ """)
+
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ '{"__classname__":"datetime.timedelta","__version__":'
+ || '2' ||
+ ',"__data__":' || CAST(interval AS TEXT) || '}'
+ """)
+
+
+def downgrade():
+ """Revert deadline interval back to float."""
+ conn = op.get_bind()
+ dialect = conn.dialect.name
+
+ if context.is_offline_mode():
+ print(
+ """
+ Manual downgrade required:
+
+ PostgreSQL:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END;
+
+ MySQL:
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END;
+
+ SQLite:
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END;
+ """
+ )
+ return
+
+ if dialect == "postgresql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN interval::jsonb ? '__data__'
+ THEN to_json((interval->>'__data__')::double precision)
+ ELSE to_json((interval::text)::double precision)
+ END
+ """)
+
+ elif dialect == "mysql":
+ op.execute("""
+ UPDATE deadline_alert
+ SET `interval` =
+ CASE
+ WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
+ THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DECIMAL(20,6))
+ ELSE CAST(`interval` AS DECIMAL(20,6))
+ END
+ """)
+
+ # Serialized VariableInterval objects do not contain a numeric "__data__" field
+ # and therefore cannot be converted back to a float representation.
+ # During downgrade, only timedelta-style serialized values are converted.
+ # Other serialized interval types (e.g. VariableInterval) are not explicitly
+ # handled here and will fall through to the generic casting branch, which may
+ # result in NULL or backend-specific casting behavior.
+ else:
+ # Detect availability of SQLite JSON functions (JSON1 extension).
+ # If available, use json_extract for robust parsing.
+ # Otherwise, fall back to string-based extraction.
+ json_functions_available = False
+ try:
+ conn.execute(sa.text("SELECT JSON_SET('{}', '$.test', 'value')")).fetchone()
+ json_functions_available = True
+ print("SQLite JSON functions detected, using optimized SQL approach")
+ except Exception:
+ print("SQLite JSON functions not available, using Python fallback for JSON processing")
+
+ if json_functions_available:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN json_extract(interval, '$.__data__') IS NOT NULL
+ THEN CAST(json_extract(interval, '$.__data__') AS REAL)
+ ELSE CAST(interval AS REAL)
+ END
+ """)
+ else:
+ op.execute("""
+ UPDATE deadline_alert
+ SET interval =
+ CASE
+ WHEN instr(interval, '__data__') > 0
+ THEN CAST(
+ substr(
+ interval,
+ instr(interval, '__data__') +
+ instr(substr(interval, instr(interval, '__data__')), ':')
+ ) AS FLOAT
+ )
+ ELSE CAST(interval AS FLOAT)
+ END
+ """)
+
+ with op.batch_alter_table("deadline_alert") as batch_op:
+ if dialect == "postgresql":
+ batch_op.alter_column(
+ "interval",
+ existing_type=sa.JSON(),
+ type_=sa.FLOAT(),
+ postgresql_using="(interval->>'__data__')::double precision",
+ existing_nullable=False,
+ )
Review Comment:
Similar to the above, the following logic has been applied here:
```
postgresql_using="""
CASE
WHEN jsonb_typeof(interval::jsonb) = 'number'
THEN interval::text::double precision
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
THEN (interval->>'__data__')::double precision
ELSE NULL
END
"""
```
Only serialized `timedelta` objects are being normalized to interval values now. Serialized `VariableInterval` objects default to `NULL`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3230156695
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
My mistake then, thanks for checking and pushing back. Sorry about that
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221220227
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
> Is it? I still see `self.interval` here, have you confirmed that this works as written?
I believe this will only apply to line 64 below as the condition will check that `self.interval` is a `timedelta` object before applying `total_seconds`. Line 60-61 is for floats and integers that have not been converted to timedelta objects.
I have tested the following DAG shape with both timedelta and the new VariableInterval object and can verify that it is working as expected:
```
def long_running_task ():
print ("Starting long-running task")
time.sleep (120)
print ("Ending long-running task")
with DAG(
dag_id="deadline_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval= [VariableInterval (key = 'deadline_seconds') / timedelta (seconds = 60],
# deadline_seconds is set to 60.
callback=AsyncCallback(
custom_alert_callback
),
),
):
PythonOperator(task_id="long_running_task",
python_callable= long_running_task)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3726122284
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -41,6 +41,7 @@
from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.models.tasklog import LogTemplate
+from airflow.sdk.definitions.deadline import VariableInterval
Review Comment:
Improvement to our dev tooling to avoid such cases: https://github.com/apache/airflow/pull/71212
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3038039760
##########
airflow-core/src/airflow/serialization/encoders.py:
##########
@@ -210,9 +210,22 @@ def encode_deadline_alert(d: DeadlineAlert | SerializedDeadlineAlert) -> dict[st
"""
from airflow.sdk.serde import serialize
+ if isinstance(d, dict):
+ return d
+
+ interval = d.interval
+
+ # Resolve dynamic interval providers (e.g. VariableInterval)
+ resolve = getattr(interval, "resolve", None)
+ if callable(resolve):
+ interval = resolve()
+
Review Comment:
getattr(interval, "resolve", None) will call .resolve() on any object that has it, not just VariableInterval. For example, if someone accidentally passes an object with an unrelated .resolve() that returns a string, the encoder calls it and then crashes later with a confusing error. We can use a Protocol so the contract is explicit and caught at lint time.
```
# With Protocol β explicit contract
from typing import Protocol
class ResolvableInterval(Protocol):
def resolve(self) -> timedelta: ...
# Now you can use it in type annotations:
def __init__(self, interval: timedelta | ResolvableInterval): ...
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3112082724
##########
airflow-core/src/airflow/serialization/encoders.py:
##########
@@ -210,9 +210,22 @@ def encode_deadline_alert(d: DeadlineAlert | SerializedDeadlineAlert) -> dict[st
"""
from airflow.sdk.serde import serialize
+ if isinstance(d, dict):
+ return d
+
+ interval = d.interval
+
+ # Resolve dynamic interval providers (e.g. VariableInterval)
+ resolve = getattr(interval, "resolve", None)
+ if callable(resolve):
+ interval = resolve()
+
Review Comment:
> getattr(interval, "resolve", None) will call .resolve() on any object that has it, not just VariableInterval. For example, if someone accidentally passes an object with an unrelated .resolve() that returns a string, the encoder calls it and then crashes later with a confusing error. We can use a Protocol so the contract is explicit and caught at lint time.
>
> ```
> # With Protocol β explicit contract
> from typing import Protocol
>
> class ResolvableInterval(Protocol):
> def resolve(self) -> timedelta: ...
>
> # Now you can use it in type annotations:
> def __init__(self, interval: timedelta | ResolvableInterval): ...
> ```
The interval is now being checked like this before deadline evaluation under the new approach. Please see the below:
```
if isinstance(interval, VariableInterval):
interval = interval.resolve()
```
The use of a Protocol is a good idea if we want to introduce other means of configuring intervals besides Variables later on but I believe an object type check is sufficient for now..
Let me know what you think.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3112082724
##########
airflow-core/src/airflow/serialization/encoders.py:
##########
@@ -210,9 +210,22 @@ def encode_deadline_alert(d: DeadlineAlert | SerializedDeadlineAlert) -> dict[st
"""
from airflow.sdk.serde import serialize
+ if isinstance(d, dict):
+ return d
+
+ interval = d.interval
+
+ # Resolve dynamic interval providers (e.g. VariableInterval)
+ resolve = getattr(interval, "resolve", None)
+ if callable(resolve):
+ interval = resolve()
+
Review Comment:
> getattr(interval, "resolve", None) will call .resolve() on any object that has it, not just VariableInterval. For example, if someone accidentally passes an object with an unrelated .resolve() that returns a string, the encoder calls it and then crashes later with a confusing error. We can use a Protocol so the contract is explicit and caught at lint time.
>
> ```
> # With Protocol β explicit contract
> from typing import Protocol
>
> class ResolvableInterval(Protocol):
> def resolve(self) -> timedelta: ...
>
> # Now you can use it in type annotations:
> def __init__(self, interval: timedelta | ResolvableInterval): ...
> ```
The interval is now being checked like this before deadline evaluation. Please see the below:
```
if isinstance(interval, VariableInterval):
interval = interval.resolve()
```
The use of a Protocol is a good idea if we want to introduce other means of configuring intervals besides Variables later on but I believe an object type check is sufficient for now..
Let me know what you think.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3114494538
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
Review Comment:
Yes. I have run multiple tests with `VariableInterval` and `timedelta` during developlment. I can do another test run and post the results if you want.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4433227637
@feruzzi
I believe I have addressed your last comment. If you are satisfied, can this be approved and merged?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221220227
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
> Is it? I still see `self.interval` here, have you confirmed that this works as written?
I believe this will only apply to line 64 below as the condition will check that `self.interval` is a `timedelta` object before applying `total_seconds`. Line 60-61 is for floats and integers that have not been converted to timedelta objects.
I have tested the following DAG shape with interval being passed using both `timedelta` and the new `VariableInterval` object and can verify that it is working as expected:
```
def long_running_task ():
print ("Starting long-running task")
time.sleep (120)
print ("Ending long-running task")
with DAG(
dag_id="deadline_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval= [VariableInterval (key = 'deadline_seconds') / timedelta (seconds = 60],
# deadline_seconds is set to 60.
callback=AsyncCallback(
custom_alert_callback
),
),
):
PythonOperator(task_id="long_running_task",
python_callable= long_running_task)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3221220227
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
Review Comment:
> Is it? I still see `self.interval` here, have you confirmed that this works as written?
I believe this will only apply to line 64 below as the condition will check that that self.interval is a timedelta object before applying `total_seconds`. Line 60-61 is for floats and integers that have not been converted to timedelta objects.
I have tested the following DAG shape with both timedelta and the new VariableInterval object and can verify that it is working as expected:
```
def long_running_task ():
print ("Starting long-running task")
time.sleep (120)
print ("Ending long-running task")
with DAG(
dag_id="deadline_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval= [VariableInterval (key = 'deadline_seconds') / timedelta (seconds = 60],
# deadline_seconds is set to 60.
callback=AsyncCallback(
custom_alert_callback
),
),
):
PythonOperator(task_id="long_running_task",
python_callable= long_running_task)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #64751:
URL: https://github.com/apache/airflow/pull/64751#issuecomment-4300311549
Quick follow-up to the triage comment above β one clarification on the "Unresolved review comments" item:
Once you believe a thread has been addressed β whether by pushing a fix, or by replying in-thread with an explanation of why the suggestion doesn't apply β **please mark the thread as resolved yourself** by clicking the "Resolve conversation" button at the bottom of each thread. Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as "still waiting on the author" and keeps the PR from moving forward. The author doing the resolve-click is the expected convention on this project.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3119707874
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
+
+ ------
+ Usage:
+ ------
+
+ .. code-block:: python
+
+ from airflow.sdk import DAG, DeadlineAlert, DeadlineReference, AsyncCallback
+
+ DAG(
+ dag_id="dag_with_variable_interval",
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=VariableInterval("deadline_minutes"),
+ callback=AsyncCallback(my_callback),
+ ),
+ )
+
+ ------
+ Notes:
+ ------
+ * Resolution occurs when deadlines are evaluated (during DagRun creation).
+ * Changes to the Variable affect only newly parsed DAGs and future DagRuns.
+ * Existing deadlines are not retroactively updated.
+ """
+
+ key: str
+
+ def resolve(self) -> timedelta:
+ try:
+ value = Variable.get(self.key)
+ except KeyError as e:
+ raise ValueError(f"VariableInterval '{self.key}' not found") from e
+
Review Comment:
Good eye.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3119700780
##########
task-sdk/src/airflow/sdk/definitions/deadline.py:
##########
@@ -340,3 +343,58 @@ def decorator(
return reference_class
return decorator
+
+
+@attrs.define(frozen=True)
+class VariableInterval:
+ """
+ Interval backed by an Airflow Variable.
+
+ This allows DeadlineAlert intervals to be configured dynamically using
+ Airflow Variables. The variable value is interpreted as minutes and
+ converted into a ``timedelta``.
Review Comment:
My opinion is that if it returns seconds, then it's easy to convert to minutes but the opposite is lossy, so we should return seconds here.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Deadline Alerts: Add dynamic interval resolution support via Variables [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #64751:
URL: https://github.com/apache/airflow/pull/64751#discussion_r3808097308
##########
airflow-core/src/airflow/models/deadline_alert.py:
##########
@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
- interval: Mapped[float] = mapped_column(Float, nullable=False)
+ interval: Mapped[dict] = mapped_column(JSON, nullable=False)
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
def __repr__(self):
- interval_seconds = int(self.interval)
- if interval_seconds >= 3600:
+ interval_seconds = None
+
+ if isinstance(self.interval, (int, float)):
+ interval_seconds = int(self.interval)
+
+ elif isinstance(self.interval, datetime.timedelta):
+ interval_seconds = int(self.interval.total_seconds())
+
+ if interval_seconds is None:
+ interval_display = "dynamic"
+ elif interval_seconds >= 3600:
Review Comment:
@feruzzi
If you check the current `main` branch, you can see this has already been addressed:
```
def __repr__(self):
interval_seconds = None
# Legacy rows store a bare number instead of a serialized dict.
if isinstance(self.interval, (int, float)):
interval_seconds = int(self.interval)
elif isinstance(self.interval, dict):
data = self.interval.get("__data__")
if isinstance(data, (int, float)):
interval_seconds = int(data)
if interval_seconds is None:
interval_display = "dynamic"
elif interval_seconds >= 3600:
interval_display = f"{interval_seconds // 3600}h"
elif interval_seconds >= 60:
interval_display = f"{interval_seconds // 60}m"
else:
interval_display = f"{interval_seconds}s"
return (
f"[DeadlineAlert] "
f"id={str(self.id)[:8]}, "
f"created_at={self.created_at}, "
f"name={self.name or 'Unnamed'}, "
f"reference={self.reference}, "
f"interval={interval_display}, "
f"callback={self.callback_def}"
)
```
The serialized timedelta representation is now handled directly by reading its numeric `__data__` value, while dynamic or unexpected serialized values fall back to dynamic. This also avoids needing to deserialize the interval or import the SDK `VariableInterval` type in `__repr__`.
The associated test has also been updated with a test case to handle scenarios where interval is in the form of a `dict`. Pleasse see the below:
```
@pytest.mark.parametrize(
("interval", "expected"),
[
# Post-0117 shape: interval is the serialized dict, not a bare number.
pytest.param(
{"__classname__": "datetime.timedelta", "__data__": 7200.0}, "interval=2h", id="timedelta_2h"
),
# A corrupted dict without ``__data__`` must still render (no raise) as dynamic.
pytest.param({"unexpected": "shape"}, "interval=dynamic", id="corrupted_dict_dynamic"),
# A VariableInterval serializes with a dict ``__data__`` (its key), not a number,
# so it renders as dynamic rather than a fixed duration.
pytest.param(
{
"__classname__": "airflow.sdk.definitions.deadline.VariableInterval",
"__data__": {"key": "deadline_seconds"},
},
"interval=dynamic",
id="variable_interval_dynamic",
),
],
)
def test_deadline_alert_repr_does_not_raise_on_json_dict_interval(
self, deadline_alert_orm, interval, expected
):
"""``DeadlineAlert.__repr__`` must not raise for the production JSON-dict interval shape."""
deadline_alert_orm.interval = interval
repr_str = repr(deadline_alert_orm) # must not raise
assert "[DeadlineAlert]" in repr_str
assert expected in repr_str
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org