Skip to content

Commit 9d44eba

Browse files
author
Sameer Mesiah
committed
Allow deadline intervals to be configured via Airflow Variables by supporting
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.
1 parent eab2938 commit 9d44eba

16 files changed

Lines changed: 573 additions & 18 deletions

File tree

‎airflow-core/docs/howto/deadline-alerts.rst‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ Creating a Deadline Alert
4242
Creating a Deadline Alert requires three mandatory parameters:
4343

4444
* Reference: When to start counting from
45-
* Interval: How far before or after the reference point to trigger the alert
45+
* Interval: How far before or after the reference point to trigger the alert (either a timedelta or a dynamic interval such as VariableInterval)
4646
* Callback: A Callback object which contains a path to a callable and optional kwargs to pass to it if the deadline is exceeded
4747

4848
Here is how Deadlines are calculated:

‎airflow-core/docs/migrations-ref.rst‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
3939
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
4040
| Revision ID | Revises ID | Airflow Version | Description |
4141
+=========================+==================+===================+==============================================================+
42-
| ``a7f3b2c1d4e5`` (head) | ``b8f3e4a1d2c9`` | ``3.3.0`` | Add allow_producer_teams column to |
42+
| ``cd8ec21786b8`` (head) | ``a7f3b2c1d4e5`` | ``3.3.0`` | Change Deadline interval to JSON. |
43+
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
44+
| ``a7f3b2c1d4e5`` | ``b8f3e4a1d2c9`` | ``3.3.0`` | Add allow_producer_teams column to |
4345
| | | | dag_schedule_asset_reference table. |
4446
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
4547
| ``b8f3e4a1d2c9`` | ``fde9ed84d07b`` | ``3.3.0`` | Add retry_delay_override and retry_reason to task_instance. |

‎airflow-core/newsfragments/64751.feature.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Allow DeadlineAlert intervals to be dynamically resolved at Deadline evaluation using objects such as VariableInterval.

‎airflow-core/src/airflow/migrations/versions/0115_3_3_0_change_deadline_interval_to_json.py‎

Lines changed: 305 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,305 @@
1+
#
2+
# Licensed to the Apache Software Foundation (ASF) under one
3+
# or more contributor license agreements. See the NOTICE file
4+
# distributed with this work for additional information
5+
# regarding copyright ownership. The ASF licenses this file
6+
# to you under the Apache License, Version 2.0 (the
7+
# "License"); you may not use this file except in compliance
8+
# with the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing,
13+
# software distributed under the License is distributed on an
14+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
# KIND, either express or implied. See the License for the
16+
# specific language governing permissions and limitations
17+
# under the License.
18+
19+
"""
20+
Change Deadline interval to JSON.
21+
22+
Revision ID: cd8ec21786b8
23+
Revises: a7f3b2c1d4e5
24+
Create Date: 2026-05-09 12:56:40.718130
25+
26+
"""
27+
28+
from __future__ import annotations
29+
30+
import sqlalchemy as sa
31+
from alembic import context, op
32+
33+
# revision identifiers, used by Alembic.
34+
revision = "cd8ec21786b8"
35+
down_revision = "a7f3b2c1d4e5"
36+
branch_labels = None
37+
depends_on = None
38+
airflow_version = "3.3.0"
39+
40+
41+
def upgrade():
42+
"""Apply change deadline interval to JSON."""
43+
conn = op.get_bind()
44+
dialect = conn.dialect.name
45+
46+
if context.is_offline_mode():
47+
print(
48+
"""
49+
Manual conversion required:
50+
51+
PostgreSQL:
52+
53+
Step 1: Convert column type.
54+
ALTER TABLE deadline_alert
55+
ALTER COLUMN interval TYPE JSONB
56+
USING to_json(interval);
57+
58+
Step 2: Convert values.
59+
UPDATE deadline_alert
60+
SET interval = json_build_object(
61+
'__classname__', 'datetime.timedelta',
62+
'__version__', 2,
63+
'__data__', (interval::text)::float
64+
)
65+
WHERE jsonb_typeof(interval::jsonb) = 'number';
66+
67+
MySQL:
68+
69+
Step 1: Convert column type.
70+
ALTER TABLE deadline_alert MODIFY COLUMN `interval` JSON;
71+
72+
Step 2: Convert values
73+
UPDATE deadline_alert
74+
SET `interval` = JSON_OBJECT(
75+
'__classname__', 'datetime.timedelta',
76+
'__version__', 2,
77+
'__data__', `interval`
78+
);
79+
80+
SQLite:
81+
82+
UPDATE deadline_alert
83+
SET interval =
84+
'{"__classname__":"datetime.timedelta","__version__":2,"__data__":'
85+
|| CAST(interval AS TEXT) || '}';
86+
"""
87+
)
88+
return
89+
90+
with op.batch_alter_table("deadline_alert") as batch_op:
91+
if dialect == "postgresql":
92+
batch_op.alter_column(
93+
"interval",
94+
existing_type=sa.FLOAT(),
95+
type_=sa.JSON(),
96+
postgresql_using="to_json(interval)",
97+
existing_nullable=False,
98+
)
99+
else:
100+
batch_op.alter_column(
101+
"interval",
102+
existing_type=sa.FLOAT(),
103+
type_=sa.JSON(),
104+
existing_nullable=False,
105+
)
106+
107+
if dialect == "postgresql":
108+
op.execute("""
109+
UPDATE deadline_alert
110+
SET interval = json_build_object(
111+
'__classname__', 'datetime.timedelta',
112+
'__version__', 2,
113+
'__data__', (interval::text)::float
114+
)
115+
WHERE jsonb_typeof(interval::jsonb) = 'number'
116+
""")
117+
118+
elif dialect == "mysql":
119+
op.execute("""
120+
UPDATE deadline_alert
121+
SET `interval` = JSON_OBJECT(
122+
'__classname__', 'datetime.timedelta',
123+
'__version__', 2,
124+
'__data__', `interval`
125+
)
126+
""")
127+
128+
else:
129+
op.execute("""
130+
UPDATE deadline_alert
131+
SET interval =
132+
'{"__classname__":"datetime.timedelta","__version__":'
133+
|| '2' ||
134+
',"__data__":' || CAST(interval AS TEXT) || '}'
135+
""")
136+
137+
138+
def downgrade():
139+
"""Revert deadline interval back to float."""
140+
conn = op.get_bind()
141+
dialect = conn.dialect.name
142+
143+
if context.is_offline_mode():
144+
print(
145+
"""
146+
Manual downgrade required:
147+
148+
PostgreSQL:
149+
150+
Step 1: Convert values.
151+
UPDATE deadline_alert
152+
SET interval =
153+
CASE
154+
WHEN jsonb_typeof(interval::jsonb) = 'number'
155+
THEN interval
156+
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
157+
THEN to_json((interval->>'__data__')::double precision)
158+
ELSE NULL
159+
END;
160+
161+
Step 2: Convert column type.
162+
ALTER TABLE deadline_alert
163+
ALTER COLUMN interval TYPE DOUBLE PRECISION
164+
USING (
165+
CASE
166+
WHEN jsonb_typeof(interval::jsonb) = 'number'
167+
THEN interval::text::double precision
168+
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
169+
THEN (interval->>'__data__')::double precision
170+
ELSE NULL
171+
END
172+
);
173+
174+
MySQL:
175+
176+
Step 1: Convert values
177+
UPDATE deadline_alert
178+
SET `interval` =
179+
CASE
180+
WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
181+
THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DOUBLE)
182+
WHEN JSON_EXTRACT(`interval`, '$.__classname__') IS NULL
183+
THEN CAST(`interval` AS DOUBLE)
184+
ELSE NULL
185+
END;
186+
187+
Step 2: Convert column type
188+
ALTER TABLE deadline_alert
189+
MODIFY COLUMN `interval` DOUBLE;
190+
191+
SQLite:
192+
193+
Step 1: Convert values
194+
UPDATE deadline_alert
195+
SET interval =
196+
CASE
197+
WHEN json_extract(interval, '$.__data__') IS NOT NULL
198+
THEN CAST(json_extract(interval, '$.__data__') AS REAL)
199+
WHEN json_extract(interval, '$.__classname__') IS NULL
200+
THEN CAST(interval AS REAL)
201+
ELSE NULL
202+
END;
203+
204+
Step 2: SQLite does not support ALTER COLUMN TYPE.
205+
Recreate the table with interval as REAL and copy data.
206+
"""
207+
)
208+
return
209+
210+
if dialect == "postgresql":
211+
op.execute("""
212+
UPDATE deadline_alert
213+
SET interval =
214+
CASE
215+
WHEN jsonb_typeof(interval::jsonb) = 'number'
216+
THEN interval
217+
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
218+
THEN to_json((interval->>'__data__')::double precision)
219+
ELSE NULL
220+
END
221+
""")
222+
223+
elif dialect == "mysql":
224+
op.execute("""
225+
UPDATE deadline_alert
226+
SET `interval` =
227+
CASE
228+
WHEN JSON_EXTRACT(`interval`, '$.__data__') IS NOT NULL
229+
THEN CAST(JSON_EXTRACT(`interval`, '$.__data__') AS DOUBLE)
230+
WHEN JSON_EXTRACT(`interval`, '$.__classname__') IS NULL
231+
THEN CAST(`interval` AS DOUBLE)
232+
ELSE NULL
233+
END
234+
""")
235+
236+
# Serialized VariableInterval objects do not contain a numeric "__data__" field
237+
# and therefore cannot be converted back to a float representation.
238+
# During downgrade, only timedelta-style serialized values are converted.
239+
# Other serialized interval types (e.g. VariableInterval) will cast as null.
240+
else:
241+
# Detect availability of SQLite JSON functions (JSON1 extension).
242+
json_functions_available = False
243+
try:
244+
conn.execute(sa.text("SELECT json_extract('{\"a\":1}', '$.a')")).fetchone()
245+
json_functions_available = True
246+
except Exception:
247+
print("SQLite JSON functions not available, using string parsing as fallback.")
248+
249+
if json_functions_available:
250+
op.execute("""
251+
UPDATE deadline_alert
252+
SET interval =
253+
CASE
254+
WHEN json_extract(interval, '$.__data__') IS NOT NULL
255+
THEN CAST(json_extract(interval, '$.__data__') AS REAL)
256+
WHEN json_extract(interval, '$.__classname__') IS NULL
257+
THEN CAST(interval AS REAL)
258+
ELSE NULL
259+
END
260+
""")
261+
else:
262+
# NOTE: This is a best-effort fallback for environments without JSON1.
263+
# It assumes a stable JSON format and may not work for all serialized values.
264+
op.execute("""
265+
UPDATE deadline_alert
266+
SET interval =
267+
CASE
268+
WHEN instr(interval, '__data__') > 0
269+
THEN CAST(
270+
substr(
271+
interval,
272+
instr(interval, '__data__') +
273+
instr(substr(interval, instr(interval, '__data__')), ':')
274+
) AS FLOAT
275+
)
276+
WHEN instr(interval, '__classname__') = 0
277+
THEN CAST(interval AS FLOAT)
278+
ELSE NULL
279+
END
280+
""")
281+
282+
with op.batch_alter_table("deadline_alert") as batch_op:
283+
if dialect == "postgresql":
284+
batch_op.alter_column(
285+
"interval",
286+
existing_type=sa.JSON(),
287+
type_=sa.FLOAT(),
288+
postgresql_using="""
289+
CASE
290+
WHEN jsonb_typeof(interval::jsonb) = 'number'
291+
THEN interval::text::double precision
292+
WHEN (interval::jsonb)->>'__classname__' = 'datetime.timedelta'
293+
THEN (interval->>'__data__')::double precision
294+
ELSE NULL
295+
END
296+
""",
297+
existing_nullable=False,
298+
)
299+
else:
300+
batch_op.alter_column(
301+
"interval",
302+
existing_type=sa.JSON(),
303+
type_=sa.FLOAT(),
304+
existing_nullable=False,
305+
)

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
from uuid import UUID
2222

2323
import uuid6
24-
from sqlalchemy import JSON, Float, ForeignKey, String, Text, Uuid, select
24+
from sqlalchemy import JSON, ForeignKey, String, Text, Uuid, select
2525
from sqlalchemy.exc import NoResultFound
2626
from sqlalchemy.orm import Mapped, mapped_column
2727

@@ -50,13 +50,22 @@ class DeadlineAlert(Base):
5050
name: Mapped[str | None] = mapped_column(String(250), nullable=True)
5151
description: Mapped[str | None] = mapped_column(Text, nullable=True)
5252
reference: Mapped[dict] = mapped_column(JSON, nullable=False)
53-
interval: Mapped[float] = mapped_column(Float, nullable=False)
53+
interval: Mapped[dict] = mapped_column(JSON, nullable=False)
5454
callback_def: Mapped[dict] = mapped_column(JSON, nullable=False)
5555

5656
def __repr__(self):
57-
interval_seconds = int(self.interval)
5857

59-
if interval_seconds >= 3600:
58+
interval_seconds = None
59+
60+
if isinstance(self.interval, (int, float)):
61+
interval_seconds = int(self.interval)
62+
63+
elif isinstance(self.interval, datetime.timedelta):
64+
interval_seconds = int(self.interval.total_seconds())
65+
66+
if interval_seconds is None:
67+
interval_display = "dynamic"
68+
elif interval_seconds >= 3600:
6069
interval_display = f"{interval_seconds // 3600}h"
6170
elif interval_seconds >= 60:
6271
interval_display = f"{interval_seconds // 60}m"

0 commit comments

Comments
 (0)