This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f840a821380 Make deadline reads and serialization robust to
dynamic/malformed intervals (#68919)
f840a821380 is described below
commit f840a8213801731028a054e733801a6449f3297d
Author: Sean Ghaeli <[email protected]>
AuthorDate: Tue Jul 28 06:53:46 2026 -0700
Make deadline reads and serialization robust to dynamic/malformed intervals
(#68919)
* Make deadline reads and serialization robust to dynamic/malformed
intervals
Hardens the read and (de)serialization paths for deadline alerts so dynamic
(``VariableInterval``) and malformed stored data no longer break the UI/API.
- UI deadline-alert response: ``DeadlineAlert.interval`` is a JSON column
holding
the Airflow-serialized interval, not a plain number. Coerce it to seconds
for a
fixed ``timedelta`` and to ``None`` for a dynamic ``VariableInterval``
(resolved
later by the scheduler), instead of letting Pydantic 500 on the dict. The
``interval`` field becomes ``float | None``.
- Drop ``interval`` from the sortable columns of the deadline-alerts
endpoint:
ordering by a JSON column sorts by structure/text, not duration, so the
result
was arbitrary and misleading.
- Deserialization: route by the encoder-stamped ``__class_path`` ahead of
the
``reference_type`` name (a custom reference may share a class name with a
builtin), and raise a clear error for a reference with no importable
``__class_path`` instead of an opaque ``KeyError``.
- ``Deadline.__repr__`` / ``DeadlineAlert.__repr__`` no longer raise: guard
the
``dagrun`` relationship (the FK can be set while the relationship is None
after a
cascade delete) and handle the dict-shaped JSON interval. A ``__repr__``
must
never raise.
- ``prune_deadlines`` explicitly excludes deadlines already marked
``missed`` so a
missed deadline (whose callback is owned by the scheduler/triggerer) and
its
queued callback are never cascade-deleted.
Generated-by: Claude Code (Opus via Claude Code) on behalf of Sean Ghaeli
* Remove redundant comments in deadline sort/deserialize
The sort-key comment restated a constraint already covered by
test_order_by_interval_is_rejected; the __class_path comment restated the
ValueError message right below it.
Generated-by: Claude Code (Opus)
* Trim PR to the UI deadlineAlerts 500 fix only
Per review: keep only the interval coercion, order_by removal, and their
tests; move repr guards, decoder routing, and prune guard to a follow-up.
---------
Co-authored-by: Sean Ghaeli <[email protected]>
---
.../api_fastapi/core_api/datamodels/ui/deadline.py | 31 +++++++-
.../api_fastapi/core_api/openapi/_private_ui.yaml | 14 ++--
.../api_fastapi/core_api/routes/ui/deadlines.py | 2 +-
.../core_api/routes/ui/test_deadlines.py | 83 +++++++++++++++++++++-
4 files changed, 120 insertions(+), 10 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
index 6f9402f2360..9c09176fea4 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py
@@ -19,9 +19,10 @@ from __future__ import annotations
from collections.abc import Iterable
from datetime import datetime
+from typing import Any
from uuid import UUID
-from pydantic import AliasPath, Field
+from pydantic import AliasPath, Field, field_validator
from airflow.api_fastapi.core_api.base import BaseModel
@@ -52,9 +53,35 @@ class DeadlineAlertResponse(BaseModel):
id: UUID
name: str | None = None
reference_type: str = Field(validation_alias=AliasPath("reference",
"reference_type"))
- interval: float = Field(description="Interval in seconds between deadline
evaluations.")
+ interval: float | None = Field(
+ default=None,
+ description=(
+ "Interval in seconds between the reference time and the deadline. "
+ "Null for a dynamic interval (e.g. a VariableInterval) whose value
is "
+ "only resolved at scheduler evaluation time."
+ ),
+ )
created_at: datetime
+ @field_validator("interval", mode="before")
+ @classmethod
+ def coerce_interval_to_seconds(cls, value: Any) -> float | None:
+ """
+ Coerce the stored ``interval`` into seconds.
+
+ ``interval`` is the Airflow-serialized SDK interval: a dict
+ ``{"__classname__": ..., "__data__": <seconds|dict>}``, not a plain
number.
+ Return the seconds for a fixed ``timedelta``, or ``None`` for a dynamic
+ interval (resolved later by the scheduler). Without this, Pydantic
500s on the dict.
+ """
+ if value is None or isinstance(value, (int, float)):
+ return value
+ if isinstance(value, dict):
+ data = value.get("__data__")
+ if isinstance(data, (int, float)):
+ return float(data)
+ return None
+
class DeadlineAlertCollectionResponse(BaseModel):
"""DeadlineAlert Collection serializer for responses."""
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index f16b2bbc9fe..75980eefb7b 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -937,13 +937,12 @@ paths:
type: string
description: 'Attributes to order by, multi criteria sort is
supported.
Prefix with `-` for descending order. Supported attributes: `id,
created_at,
- name, interval`'
+ name`'
default:
- created_at
title: Order By
description: 'Attributes to order by, multi criteria sort is
supported. Prefix
- with `-` for descending order. Supported attributes: `id,
created_at, name,
- interval`'
+ with `-` for descending order. Supported attributes: `id,
created_at, name`'
responses:
'200':
description: Successful Response
@@ -2849,9 +2848,13 @@ components:
type: string
title: Reference Type
interval:
- type: number
+ anyOf:
+ - type: number
+ - type: 'null'
title: Interval
- description: Interval in seconds between deadline evaluations.
+ description: Interval in seconds between the reference time and the
deadline.
+ Null for a dynamic interval (e.g. a VariableInterval) whose value
is only
+ resolved at scheduler evaluation time.
created_at:
type: string
format: date-time
@@ -2860,7 +2863,6 @@ components:
required:
- id
- reference_type
- - interval
- created_at
title: DeadlineAlertResponse
description: DeadlineAlert serializer for responses.
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
index 06eda42ed89..2d080609c35 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
@@ -166,7 +166,7 @@ def get_dag_deadline_alerts(
SortParam,
Depends(
SortParam(
- ["id", "created_at", "name", "interval"],
+ ["id", "created_at", "name"],
DeadlineAlert,
).dynamic_depends(default="created_at")
),
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
index acadab3a6b1..6c83044e17c 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_deadlines.py
@@ -17,6 +17,8 @@
from __future__ import annotations
+from datetime import timedelta
+
import pytest
from sqlalchemy import select
@@ -26,7 +28,7 @@ from airflow.models.deadline_alert import DeadlineAlert
from airflow.models.serialized_dag import SerializedDagModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk.definitions.callback import AsyncCallback
-from airflow.sdk.definitions.deadline import DeadlineReference
+from airflow.sdk.definitions.deadline import DeadlineReference,
VariableInterval
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -507,6 +509,17 @@ class TestGetDagDeadlineAlerts:
response = test_client.get("/dags/nonexistent_dag/deadlineAlerts")
assert response.status_code == 404
+ @pytest.mark.parametrize("order_by", ["interval", "-interval"])
+ def test_order_by_interval_is_rejected(self, test_client, order_by):
+ """``interval`` is a serialized-JSON column
(timedelta/VariableInterval dict), so DB-level
+ ordering by it is meaningless (sorts by JSON structure, not duration).
It must NOT be an
+ allowed sort key — the endpoint should reject it with a 400 rather
than silently returning
+ an arbitrarily-ordered list.
+ """
+ response = test_client.get(f"/dags/{DAG_ID}/deadlineAlerts",
params={"order_by": order_by})
+ assert response.status_code == 400
+ assert "interval" in response.json()["detail"]
+
def test_should_response_401(self, unauthenticated_test_client):
response =
unauthenticated_test_client.get(f"/dags/{DAG_ID}/deadlineAlerts")
assert response.status_code == 401
@@ -514,3 +527,71 @@ class TestGetDagDeadlineAlerts:
def test_should_response_403(self, unauthorized_test_client):
response =
unauthorized_test_client.get(f"/dags/{DAG_ID}/deadlineAlerts")
assert response.status_code == 403
+
+
+class TestDeadlineAlertsIntervalSerialization:
+ """Regression tests for the ``interval`` column shape on the
deadlineAlerts endpoint.
+
+ ``DeadlineAlert.interval`` is a JSON column that, in production, holds the
+ Airflow-*serialized* interval — ``encode_deadline_alert`` stores
+ ``serialize(self.interval)``, not a plain number. A fixed ``timedelta``
becomes
+ ``{"__classname__": "datetime.timedelta", "__version__": 2, "__data__":
<seconds>}``
+ and a dynamic ``VariableInterval`` becomes
+ ``{"__classname__": ".../VariableInterval", "__data__": {"key": ...}}``.
+
+ The response model originally typed ``interval`` as a bare ``float``, so
Pydantic
+ raised on that dict and the endpoint returned 500 — which broke the
run-page
+ ``DeadlineStatus`` badge. The existing fixtures masked this by seeding a
bare float
+ (``interval=3600.0``), a value the real creation path never produces.
These tests
+ seed the realistic serialized forms.
+ """
+
+ @pytest.fixture
+ def dag_with_serialized_intervals(self, dag_maker, session):
+ from airflow.sdk.serde import serialize
+
+ dag_id = "dag_serialized_interval"
+ with dag_maker(dag_id, serialized=True, session=session):
+ EmptyOperator(task_id="task")
+ dag_maker.sync_dagbag_to_db()
+ session.commit()
+
+ serialized_dag =
session.scalar(select(SerializedDagModel).where(SerializedDagModel.dag_id ==
dag_id))
+ # Fixed interval: stored as the serialized timedelta dict, exactly as
+ # encode_deadline_alert would persist it.
+ session.add(
+ DeadlineAlert(
+ serialized_dag_id=serialized_dag.id,
+ name="fixed_interval_alert",
+
reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
+ interval=serialize(timedelta(seconds=300)),
+ callback_def={"path": _CALLBACK_PATH},
+ )
+ )
+ # Dynamic interval: a VariableInterval serializes to a dict with no
fixed
+ # seconds — the value is only resolved at scheduler evaluation time.
+ session.add(
+ DeadlineAlert(
+ serialized_dag_id=serialized_dag.id,
+ name="dynamic_interval_alert",
+
reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
+ interval=serialize(VariableInterval("deadline_seconds")),
+ callback_def={"path": _CALLBACK_PATH},
+ )
+ )
+ session.commit()
+ return dag_id
+
+ def test_serialized_timedelta_interval_does_not_500(self, test_client,
dag_with_serialized_intervals):
+ """A fixed timedelta interval is coerced to its seconds value (not a
500)."""
+ response =
test_client.get(f"/dags/{dag_with_serialized_intervals}/deadlineAlerts")
+ assert response.status_code == 200
+ alerts = {a["name"]: a for a in response.json()["deadline_alerts"]}
+ assert alerts["fixed_interval_alert"]["interval"] == 300.0
+
+ def test_dynamic_variable_interval_serializes_as_null(self, test_client,
dag_with_serialized_intervals):
+ """A dynamic VariableInterval has no fixed seconds, so interval is
null (not a 500)."""
+ response =
test_client.get(f"/dags/{dag_with_serialized_intervals}/deadlineAlerts")
+ assert response.status_code == 200
+ alerts = {a["name"]: a for a in response.json()["deadline_alerts"]}
+ assert alerts["dynamic_interval_alert"]["interval"] is None
