Deadline Alerts: Add dynamic interval resolution support via Variables by SameerMesiah97 · Pull Request #64751 · apache/airflow

SameerMesiah97 · 2026-04-06T00:20:58Z

Description

This change allows deadline intervals to be configured dynamically via Airflow Variables by introducing support for interval objects (e.g. VariableInterval).

VariableInterval defers resolution until deadline evaluation time, where it retrieves the Variable value (interpreted as minutes) and converts it into a timedelta. This ensures that intervals are always materialized to timedelta before being used at runtime.

To support this, the deadline_alert.interval column is migrated from Float to JSON, allowing serialized interval objects to be stored instead of only numeric values.

Rationale

The primary motivation is to enable dynamic configuration of deadline intervals using Airflow Variables without requiring changes to DAG code. This allows operators to adjust deadline behavior externally, making the feature more flexible and easier to manage in production environments.

An explicit VariableInterval type is used instead of implicitly resolving Variables to avoid silent fallbacks and hidden side effects. This makes the behavior opt-in and predictable, ensuring that DAG authors are aware when interval values are dynamically sourced from Variables.

Resolving intervals at deadline evaluation time ensures that values are materialized to timedelta just before use, while remaining stable for the lifetime of a DAG Run. This approach avoids introducing additional execution-time mechanisms and keeps the implementation aligned with existing deadline evaluation behavior.

Migrations

The migration 0111_3_3_0_change_deadline_interval_to_json.py updates the deadline_alert.interval column from Float to JSON to support storing serialized interval objects. Existing numeric values are preserved and continue to be supported during decoding for backwards compatibility. Serialized VariableInterval objects are skipped during downgrade as they do not contain interval values.

Tests

Added tests covering:

VariableInterval.resolve() for valid and invalid Variable values
DAG Run behavior with VariableInterval, including:
- Stable resolution within the same DAG Run.
- Failure when the referenced Variable is missing.
- Coverage added to an existing DAGRun success test for VariableInterval.

Other tests have been updated to accommodate the serialization of the timedelta object where necessary.

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.
Added the migration to the "Reference for Database Migrations" documentation.

Backwards Compatibility

Existing timedelta-based intervals continue to work unchanged, and previously stored numeric interval values remain supported during decoding to ensure backwards compatibility. Dynamic interval support is introduced as an opt-in feature via VariableInterval, allowing users to adopt it incrementally. A database migration is required to change the interval column from Float to JSON to support serialized interval objects, but existing data remains compatible and continues to function without modification.

Closes: #63852

shivaam · 2026-04-06T05:21:34Z

This is in draft but a few thoughts on this:

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.
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.

shivaam · 2026-04-06T05:26:22Z

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

+    # Resolve dynamic interval providers (e.g. VariableInterval)
+    resolve = getattr(interval, "resolve", None)
+    if callable(resolve):
+        interval = resolve()
+


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): ...

ferruzzi · 2026-04-06T21:39:15Z

@seanghaeli may want to have an eye on this one as well.

SameerMesiah97 · 2026-04-06T21:52:04Z

@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.

…pporting interval objects (e.g. VariableInterval) that resolve at evaluation time. Migrate deadline_alert.interval from Float to JSON to support serialized interval objects. Ensure intervals are materialized to timedelta at runtime. Add unit tests for interval resolution and DAGRun tests validating deadline behavior and stability across updates.

Copilot

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds support for dynamically configured DeadlineAlert intervals via an interval object (VariableInterval) resolved at deadline evaluation time, and migrates persisted alert intervals from Float to JSON so serialized interval objects can be stored.

Changes:

Introduces VariableInterval (backed by Airflow Variables) and allows DeadlineAlert.interval to accept resolvable intervals.
Updates deadline alert serialization/deserialization to store intervals as serialized objects (with backward compatibility for numeric seconds).
Migrates deadline_alert.interval from Float to JSON, and updates docs/tests accordingly.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
task-sdk/src/airflow/sdk/definitions/deadline.py	Adds `_ResolvableInterval` protocol and `VariableInterval` implementation; broadens accepted `interval` types.
task-sdk/tests/task_sdk/definitions/test_deadline.py	Adds unit tests for `VariableInterval.resolve()` valid/invalid cases.
airflow-core/src/airflow/serialization/encoders.py	Serializes `DeadlineAlert.interval` via generic `serialize()` rather than `total_seconds()`.
airflow-core/src/airflow/serialization/decoders.py	Adds backward-compatible decoding for numeric intervals and deserializes interval objects.
airflow-core/src/airflow/serialization/definitions/dag.py	Resolves `VariableInterval` during DagRun deadline evaluation before using interval.
airflow-core/src/airflow/models/deadline_alert.py	Changes ORM column type to JSON and adjusts `__repr__` to handle non-float interval shapes.
airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py	Adds migration converting `deadline_alert.interval` from Float to JSON, with downgrade logic.
airflow-core/src/airflow/utils/db.py	Updates the `3.3.0` migration head revision mapping.
airflow-core/tests/unit/models/test_serialized_dag.py	Updates assertion for newly JSON-serialized interval representation.
airflow-core/tests/unit/models/test_dag.py	Updates interval extraction from stored alerts to handle JSON representation.
airflow-core/tests/unit/models/test_dagrun.py	Adds DagRun tests for `VariableInterval` stability and missing-variable failure; parameterizes an existing success test.
airflow-core/docs/howto/deadline-alerts.rst	Documents that `interval` can be a `timedelta` or dynamic interval object (e.g., `VariableInterval`).
airflow-core/docs/migrations-ref.rst	Adds the new migration to the migrations reference table.
airflow-core/newsfragments/64751.feature.rst	Announces dynamic interval support for deadline alerts.

Copilot · 2026-04-10T19:58:43Z

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

    return decorator
+
+
+@attrs.define


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.

Suggested change

@attrs.define

@attrs.define(unsafe_hash=True)

Copilot · 2026-04-10T19:58:44Z

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

+    """
+
+    key: str
+


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.

Suggested change

def __hash__(self) -> int:

return hash(self.key)

Copilot · 2026-04-10T19:58:44Z

airflow-core/src/airflow/models/deadline_alert.py

+        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"


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.

Copilot · 2026-04-10T19:58:44Z

airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py

+    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
+        """)


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).

Copilot · 2026-04-10T19:58:45Z

airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py

+    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,
+            )


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).

Copilot · 2026-04-10T19:58:45Z

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

+    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))


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.

Copilot · 2026-04-10T19:58:45Z

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.


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.

Copilot · 2026-04-10T19:58:45Z

airflow-core/src/airflow/migrations/versions/0111_3_3_0_change_deadline_interval_to_json.py

+    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';


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.

SameerMesiah97 requested review from amoghrajesh, ashb, bolkedebruin and kaxil as code owners April 6, 2026 00:21

boring-cyborg bot added area:DAG-processing area:deadline-alerts AIP-86 (former AIP-57) area:task-sdk kind:documentation labels Apr 6, 2026

SameerMesiah97 force-pushed the 63852-Deadline-Alerts-VariableInterval branch from 08de7dd to 7a6c579 Compare April 6, 2026 00:24

SameerMesiah97 marked this pull request as draft April 6, 2026 00:24

shivaam reviewed Apr 6, 2026

View reviewed changes

SameerMesiah97 force-pushed the 63852-Deadline-Alerts-VariableInterval branch from 7a6c579 to 1acd5ea Compare April 6, 2026 21:37

SameerMesiah97 force-pushed the 63852-Deadline-Alerts-VariableInterval branch 6 times, most recently from a700235 to 14997b7 Compare April 9, 2026 02:03

SameerMesiah97 force-pushed the 63852-Deadline-Alerts-VariableInterval branch from 14997b7 to 598ede2 Compare April 9, 2026 02:08

kaxil requested a review from Copilot April 10, 2026 19:55

Copilot AI reviewed Apr 10, 2026

View reviewed changes

		@@ -0,0 +1 @@
		Allow DeadlineAlert intervals to be dynamically resolved at DAG parse time using objects such as VariableInterval.

Conversation

SameerMesiah97 commented Apr 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shivaam commented Apr 6, 2026

Uh oh!

shivaam Apr 6, 2026

Choose a reason for hiding this comment

Uh oh!

ferruzzi commented Apr 6, 2026

Uh oh!

SameerMesiah97 commented Apr 6, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

SameerMesiah97 commented Apr 6, 2026 •

edited

Loading