This is an automated email from the ASF dual-hosted git repository.
potiuk 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 40f76fc1080 Add example DAG demonstrating Deadline Alerts (#66269)
40f76fc1080 is described below
commit 40f76fc1080ae67a24fd6d82ecc3960bbe12698a
Author: Jeongseok Kang <[email protected]>
AuthorDate: Sun Aug 30 08:28:35 2026 +0900
Add example DAG demonstrating Deadline Alerts (#66269)
* Add example DAG demonstrating Deadline Alerts
Adds airflow-core/src/airflow/example_dags/example_deadline_alert.py so
users can try the Deadline Alerts feature on a fresh Airflow install
without writing a custom DAG.
Also fixes airflow.sdk's type stub (__init__.pyi) which was missing
AsyncCallback, SyncCallback, DeadlineAlert, and DeadlineReference -
they exist at runtime via lazy imports but the stub had no entries,
causing `from airflow.sdk import DeadlineAlert` (the form used in
docs/howto/deadline-alerts.rst) to fail mypy attr-defined checks.
related: #66268
* Update airflow-core/src/airflow/example_dags/example_deadline_alert.py
Co-authored-by: hojeong park <[email protected]>
* Update airflow-core/src/airflow/example_dags/example_deadline_alert.py
Co-authored-by: hojeong park <[email protected]>
* Allow deadline UUIDs in serialized DAG schema
`SerializedDagModel.write_dag` rewrites `dag.deadline` from a
list of encoded alert dicts to a list of UUIDv7 strings before
persisting, but the JSON schema only permitted dict / list[dict] /
null. Validating any DAG with a deadline against the post-write
shape therefore raised `ValidationError`. Add `list[str]` to the
`deadline` `anyOf` so the schema matches what is actually stored.
* Make example deadline alert DAG actually trip the alert
Address PR review feedback:
- Switch the example from `DAGRUN_LOGICAL_DATE` to `DAGRUN_QUEUED_AT`.
Logical date for a manual run can be far in the past or future, so
`LOGICAL_DATE + interval` rarely produces a meaningful deadline for
a demo DAG. Queued-at is predictable and matches what a reader would
expect when triggering the example.
- Drop the interval from one hour to thirty seconds and sleep for a
minute inside the task so triggering the example DAG actually
observably crosses the deadline and fires the alert.
- Drop the `SyncCallback` stub entry from `airflow.sdk.__init__.pyi`.
The example only uses `AsyncCallback`; `SyncCallback` is mid-rollout
and out of scope for this PR.
* Add regression test for deadline UUID schema acceptance
write_dag rewrites dag.deadline from list[dict] to list[str] (UUIDv7)
before persistence (_generate_deadline_uuids), so the JSON schema's
deadline anyOf must accept list[str]. test_write_dag exposes this
transitively via re-validation of stored data; this commit adds a
focused unit test that guards the contract directly, plus a docstring
cross-reference in _generate_deadline_uuids so the schema coupling is
visible from the rewrite site.
* Add SyncCallback to airflow.sdk type stub
Runtime __init__.py exports SyncCallback via lazy import (alongside
AsyncCallback, DeadlineAlert, DeadlineReference) and the deadline-alerts
documentation imports it directly. Aligning the .pyi stub avoids mypy
attr-defined errors for users following the docs.
* Clarify why example_deadline_alert sleeps synchronously
The 60s sleep is a deliberate choice — a deferred sensor would
demonstrate the same trip path but obscure the DeadlineAlert focus.
Replace the two-line WHAT comment with one line that captures the WHY.
* Make deadline callback context parameter explicit in example DAG
* chore: Remove out-of-scope core changes
* refactor: Remove redundant import aliasing
---------
Co-authored-by: hojeong park <[email protected]>
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../airflow/example_dags/example_deadline_alert.py | 61 ++++++++++++++++++++++
task-sdk/src/airflow/sdk/__init__.pyi | 6 +++
2 files changed, 67 insertions(+)
diff --git a/airflow-core/src/airflow/example_dags/example_deadline_alert.py
b/airflow-core/src/airflow/example_dags/example_deadline_alert.py
new file mode 100644
index 00000000000..54f5f7821bb
--- /dev/null
+++ b/airflow-core/src/airflow/example_dags/example_deadline_alert.py
@@ -0,0 +1,61 @@
+# 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.
+"""Example Dag demonstrating :class:`~airflow.sdk.DeadlineAlert` usage."""
+
+from __future__ import annotations
+
+import logging
+import time
+from datetime import timedelta
+
+import pendulum
+
+# [START example_deadline_alert]
+from airflow.sdk import DAG, AsyncCallback, DeadlineAlert, DeadlineReference,
task
+
+log = logging.getLogger(__name__)
+
+
+async def notify_deadline_missed(context: dict, **kwargs):
+ """Async callback executed by the Triggerer when the deadline is missed."""
+ dag_run = context.get("dag_run")
+ log.warning("Deadline missed for dag_run=%s", dag_run)
+
+
+with DAG(
+ dag_id="example_deadline_alert",
+ description="Demonstrates DeadlineAlert with DAGRUN_QUEUED_AT reference.",
+ schedule=None,
+ start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
+ catchup=False,
+ tags=["example", "deadline"],
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(seconds=30),
+ callback=AsyncCallback(notify_deadline_missed),
+ name="example_deadline",
+ ),
+) as dag:
+
+ @task
+ def hello_deadline():
+ log.info("Hello from a Dag with a deadline!")
+ # Sleep past the deadline (sync, not a deferred sensor) to keep the
demo focused on DeadlineAlert.
+ time.sleep(60)
+
+ hello_deadline()
+# [END example_deadline_alert]
diff --git a/task-sdk/src/airflow/sdk/__init__.pyi
b/task-sdk/src/airflow/sdk/__init__.pyi
index a86bdeceefd..37585f5d796 100644
--- a/task-sdk/src/airflow/sdk/__init__.pyi
+++ b/task-sdk/src/airflow/sdk/__init__.pyi
@@ -53,6 +53,7 @@ from airflow.sdk.definitions.asset import (
from airflow.sdk.definitions.asset.access_control import AssetAccessControl as
AssetAccessControl
from airflow.sdk.definitions.asset.decorators import asset as asset
from airflow.sdk.definitions.asset.metadata import Metadata as Metadata
+from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
from airflow.sdk.definitions.connection import Connection as Connection
from airflow.sdk.definitions.context import (
Context as Context,
@@ -60,6 +61,7 @@ from airflow.sdk.definitions.context import (
get_parsing_context as get_parsing_context,
)
from airflow.sdk.definitions.dag import DAG as DAG, dag as dag
+from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference
from airflow.sdk.definitions.decorators import (
result as result,
setup as setup,
@@ -143,6 +145,7 @@ __all__ = [
"AssetAny",
"AssetOrTimeSchedule",
"AssetWatcher",
+ "AsyncCallback",
"BaseAsyncOperator",
"BaseBranchOperator",
"BaseHook",
@@ -161,6 +164,8 @@ __all__ = [
"DAG",
"DagRunState",
"DayWindow",
+ "DeadlineAlert",
+ "DeadlineReference",
"DeltaDataIntervalTimetable",
"DeltaTriggerTimetable",
"EdgeModifier",
@@ -199,6 +204,7 @@ __all__ = [
"StartOfQuarterMapper",
"StartOfWeekMapper",
"StartOfYearMapper",
+ "SyncCallback",
"TaskGroup",
"TaskInstanceState",
"TriggerRule",
