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 6f9402f23603d..9c09176fea414 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 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__": }``, 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 ca8f729998250..7128cab072ff8 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 @@ -963,13 +963,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 @@ -2859,9 +2858,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 @@ -2870,7 +2873,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 06eda42ed8957..2d080609c3569 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 acadab3a6b16f..6c83044e17c3f 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.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 @@ def test_should_response_404_for_nonexistent_dag(self, test_client): 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 @@ def test_should_response_401(self, unauthenticated_test_client): 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__": }`` + 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