You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "amoghrajesh (via GitHub)" <gi...@apache.org> on 2026/01/27 11:46:30 UTC
[PR] Split Deadline Alerts into core and task-sdk components [airflow]
amoghrajesh opened a new pull request, #61118:
URL: https://github.com/apache/airflow/pull/61118
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [x] Yes (please specify the tool below)
Used cursor IDE
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
### Why This Change?
Trying to assist the client-server separation by moving worker side deadline functionality to `task-sdk` while keeping the DB dependent evaluation logic in `airflow-core`. This follows the established pattern from the assets migration: #58993
### What stays where?
In simpler terms, this PR tries to address this.
#### SDK
- `DeadlineAlert`: Its the user facing class for defining deadline alerts (no serialization methods from now)
- `DeadlineReference`: The same factory class for creating deadline references
- `ReferenceModels.*`: Original reference implementations for backward compatibility
The principle here it to keep the lightweight DAG authoring interface with no database dependencies in sdk
#### Core / Serialization module
- `SerializedDeadlineAlert`: Internal representation for core usages used post deserialization of a DeadlineAlert
- `SerializedReferenceModels.*`: Reference implementations with database access
- `encode_deadline_alert()` / `decode_deadline_alert()`: Centralized serialization functions used to ser/deser the deadline alerts
The principle here is to keep the serialization, deserialization, and deadline evaluation with database access in core.
### Serialization Changes
#### Structure
Serialization format remains **unchanged** - no breaking changes to stored DAGs:
```json
{
"__type": "deadline_alert",
"__var": {
"reference": {"reference_type": "DagRunLogicalDateDeadline"},
"interval": 3600.0,
"callback": {
"__classname__": "airflow.sdk.definitions.callback.AsyncCallback",
"__version__": 0,
"__data__": {"path": "...", "kwargs": {...}}
}
}
}
```
Same with the flow of control
1. Encode (DAG Processor): `DeadlineAlert` → dict via `encode_deadline_alert()` using `airflow.sdk.serde`
2. Stored as JSON in database
3. Decode (Scheduler): dict → `SerializedDeadlineAlert` via `decode_deadline_alert()`
4. Evaluate: `SerializedReferenceModels` uses database session to calculate deadlines
One thing of note is the callback serialisation, I chose to continue using serde for this purpose because BaseSerialisation cannot handle callbacks. Using serde made sense since this part of serialisation runs in dag processor, which untilmately is not a _core component_ and can use task sdk. So, flow:
- Uses `airflow.sdk.serde.serialize()` / `deserialize()` for proper callback handling
- Runs in DAG Processor context where SDK is available
- Callbacks fully serialize with path and kwargs (no string representations)
### Backward Compatibility
- Serialization format identical to main branch
- Reference class names unchanged (e.g., `DagRunLogicalDateDeadline` not `SerializedDagRunLogicalDateDeadline`)
- Existing serialized DAGs deserialize correctly
- Internal API only - no user-facing changes, so nothing to worry about hopefully
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2734968924
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -131,6 +135,32 @@ def decode_asset_like(var: dict[str, Any]) -> SerializedAssetBase:
raise ValueError(f"deserialization not implemented for DAT {data_type!r}")
+def decode_deadline_alert(encoded_data: dict):
+ """
+ Decode a previously serialized deadline alert.
+
+ :meta private:
+ """
+ from datetime import timedelta
+
+ from airflow.sdk.serde import deserialize
+ from airflow.serialization.definitions.deadline import SerializedDeadlineAlert
Review Comment:
There is no reason that was done, let me make the change to make it proper
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh merged PR #61118:
URL: https://github.com/apache/airflow/pull/61118
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2732266217
##########
airflow-core/tests/unit/models/test_deadline_alert.py:
##########
@@ -175,7 +175,7 @@ def test_deadline_alert_hash(self, session, deadline_reference):
assert hash(alert1) == hash(alert2)
def test_deadline_alert_reference_class_property(self, deadline_alert_orm):
- assert deadline_alert_orm.reference_class == ReferenceModels.DagRunQueuedAtDeadline
+ assert deadline_alert_orm.reference_class == SerializedReferenceModels.DagRunQueuedAtDeadline
Review Comment:
Now reference_class returns a `SerializedReferenceModels`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2742827081
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -512,8 +513,10 @@ def serialize(
)
elif isinstance(var, DAG):
return cls._encode(DagSerialization.serialize_dag(var), type_=DAT.DAG)
- elif isinstance(var, DeadlineAlert):
- return cls._encode(DeadlineAlert.serialize_deadline_alert(var), type_=DAT.DEADLINE_ALERT)
+ elif isinstance(var, (DeadlineAlert, SerializedDeadlineAlert)):
Review Comment:
Alright, I guess it can't hurt.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2735017854
##########
airflow-core/tests/unit/serialization/test_serialized_objects.py:
##########
@@ -485,14 +485,16 @@ def test_serialize_deserialize_deadline_alert(reference):
callback=AsyncCallback(empty_callback_for_deadline, kwargs=TEST_CALLBACK_KWARGS),
)
- serialized = original.serialize_deadline_alert()
+ # Use BaseSerialization like assets do
+ serialized = BaseSerialization.serialize(original)
assert serialized[Encoding.TYPE] == DAT.DEADLINE_ALERT
assert set(serialized[Encoding.VAR].keys()) == public_deadline_alert_fields
- deserialized = DeadlineAlert.deserialize_deadline_alert(serialized)
+ deserialized = BaseSerialization.deserialize(serialized)
assert deserialized.reference.serialize_reference() == reference.serialize_reference()
assert deserialized.interval == original.interval
- assert deserialized.callback == original.callback
+ # Callback is deserialized as SerializedDeadlineAlert, which may have different callback representation
Review Comment:
It's actually supposed to be the same, this was when I was exploring some other option to support callback serialisation for callbacks. Now I am just using serde like earlier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2735003392
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -512,8 +513,10 @@ def serialize(
)
elif isinstance(var, DAG):
return cls._encode(DagSerialization.serialize_dag(var), type_=DAT.DAG)
- elif isinstance(var, DeadlineAlert):
- return cls._encode(DeadlineAlert.serialize_deadline_alert(var), type_=DAT.DEADLINE_ALERT)
+ elif isinstance(var, (DeadlineAlert, SerializedDeadlineAlert)):
Review Comment:
On normal days, `DeadlineAlert` should be the one coming in, but I followed this pattern just defensively similar to how assets does it: https://github.com/apache/airflow/pull/58993/changes#diff-2f03f28d3201c630dfbae758caf95ada61ab10328de789fc464f6e27c4afb2d0R700-R701
There is a very rare case when a serialized version can come in, like for example reserializing. If someone loads a DAG from db and tries to modify the in memory dag object. Rare case, so maybe I can remove it as well if you like.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2735024339
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -28,3 +46,235 @@ class DeadlineAlertFields:
REFERENCE = "reference"
INTERVAL = "interval"
CALLBACK = "callback"
+
+
Review Comment:
That's that yes, its mostly copypasta from the older models but with some renaming. You can just skim this one through, no need to check in great detail
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #61118:
URL: https://github.com/apache/airflow/pull/61118#issuecomment-3809215120
@ferruzzi thanks for your review, I have handled your comments now, let me know how it looks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2742822021
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -28,3 +46,235 @@ class DeadlineAlertFields:
REFERENCE = "reference"
INTERVAL = "interval"
CALLBACK = "callback"
+
+
Review Comment:
Cool, just double-checking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2734113801
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -131,6 +135,32 @@ def decode_asset_like(var: dict[str, Any]) -> SerializedAssetBase:
raise ValueError(f"deserialization not implemented for DAT {data_type!r}")
+def decode_deadline_alert(encoded_data: dict):
+ """
+ Decode a previously serialized deadline alert.
+
+ :meta private:
+ """
+ from datetime import timedelta
+
+ from airflow.sdk.serde import deserialize
+ from airflow.serialization.definitions.deadline import SerializedDeadlineAlert
Review Comment:
Here and elsewhere: Why are we introducing local imports all over the place? Isn't the project standard to use top-level imports unless we have a reason not to (circular imports, etc)?
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -512,8 +513,10 @@ def serialize(
)
elif isinstance(var, DAG):
return cls._encode(DagSerialization.serialize_dag(var), type_=DAT.DAG)
- elif isinstance(var, DeadlineAlert):
- return cls._encode(DeadlineAlert.serialize_deadline_alert(var), type_=DAT.DEADLINE_ALERT)
+ elif isinstance(var, (DeadlineAlert, SerializedDeadlineAlert)):
Review Comment:
I'm not sure I follow why it can be either one. If you are attempting to serialize then shouldn't it be the unserialized version coming in? Or is this just defensive in case we try to re-serialize? I don't see this pattern in the other elif clauses so it seems a odd enough to ask.
##########
airflow-core/tests/unit/serialization/test_serialized_objects.py:
##########
@@ -485,14 +485,16 @@ def test_serialize_deserialize_deadline_alert(reference):
callback=AsyncCallback(empty_callback_for_deadline, kwargs=TEST_CALLBACK_KWARGS),
)
- serialized = original.serialize_deadline_alert()
+ # Use BaseSerialization like assets do
+ serialized = BaseSerialization.serialize(original)
assert serialized[Encoding.TYPE] == DAT.DEADLINE_ALERT
assert set(serialized[Encoding.VAR].keys()) == public_deadline_alert_fields
- deserialized = DeadlineAlert.deserialize_deadline_alert(serialized)
+ deserialized = BaseSerialization.deserialize(serialized)
assert deserialized.reference.serialize_reference() == reference.serialize_reference()
assert deserialized.interval == original.interval
- assert deserialized.callback == original.callback
+ # Callback is deserialized as SerializedDeadlineAlert, which may have different callback representation
Review Comment:
Shouldn't serializing then deserializing return the original? Or at least "a new object containing equal values", not the literal same object.
##########
airflow-core/src/airflow/serialization/definitions/deadline.py:
##########
@@ -28,3 +46,235 @@ class DeadlineAlertFields:
REFERENCE = "reference"
INTERVAL = "interval"
CALLBACK = "callback"
+
+
Review Comment:
For the most part this is all copypasta from the old ReferenceModels, other than renaming some of the the classes and type hints? Just double checking, let me know if there are specific changes I should take a closer look at closer in this file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2735026151
##########
airflow-core/tests/unit/serialization/test_serialized_objects.py:
##########
@@ -485,14 +485,16 @@ def test_serialize_deserialize_deadline_alert(reference):
callback=AsyncCallback(empty_callback_for_deadline, kwargs=TEST_CALLBACK_KWARGS),
)
- serialized = original.serialize_deadline_alert()
+ # Use BaseSerialization like assets do
+ serialized = BaseSerialization.serialize(original)
assert serialized[Encoding.TYPE] == DAT.DEADLINE_ALERT
assert set(serialized[Encoding.VAR].keys()) == public_deadline_alert_fields
- deserialized = DeadlineAlert.deserialize_deadline_alert(serialized)
+ deserialized = BaseSerialization.deserialize(serialized)
assert deserialized.reference.serialize_reference() == reference.serialize_reference()
assert deserialized.interval == original.interval
- assert deserialized.callback == original.callback
+ # Callback is deserialized as SerializedDeadlineAlert, which may have different callback representation
Review Comment:
Reverted that change in 24548f2e96
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2732268524
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1321,7 +1322,7 @@ def test_dagrun_success_prunes_dagrun_deadlines(
self, mock_get_by_id, mock_prune, session, deadline_test_dag
):
mock_deadline_alert = mock.MagicMock()
- mock_deadline_alert.reference_class = ReferenceModels.FixedDatetimeDeadline
+ mock_deadline_alert.reference_class = SerializedReferenceModels.FixedDatetimeDeadline
Review Comment:
`SerializedReferenceModels` is returned now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #61118:
URL: https://github.com/apache/airflow/pull/61118#issuecomment-3822062006
Thanks @ferruzzi!
Just to gain some more confidence prior to merge, I tried it out functionally.
Used this custom callback:
```python
async def custom_async_callback(**kwargs):
context = kwargs.get("context", {})
print(
f"Deadline exceeded for Dag {context.get('dag_run', {}).get('dag_id')}!"
)
print(f"Context: {context}")
print(f"Alert type: {kwargs.get('alert_type')}")
```
This DAG:
```python
from datetime import timedelta
from deadline_callback import custom_async_callback
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG
from airflow.sdk.definitions.deadline import AsyncCallback, DeadlineAlert, DeadlineReference
with DAG(
dag_id="custom_deadline_alert",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(seconds=10),
callback=AsyncCallback(
custom_async_callback,
kwargs={"alert_type": "time_exceeded"},
),
),
):
BashOperator(task_id="example_task", bash_command="sleep 30")
```
And it works as I expect it to:
<img width="2559" height="1005" alt="image" src="https://github.com/user-attachments/assets/1d02859e-5f52-49c3-94e9-403a8d1c8d64" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Split Deadline Alerts into core and task-sdk components [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #61118:
URL: https://github.com/apache/airflow/pull/61118#discussion_r2735025714
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -131,6 +135,32 @@ def decode_asset_like(var: dict[str, Any]) -> SerializedAssetBase:
raise ValueError(f"deserialization not implemented for DAT {data_type!r}")
+def decode_deadline_alert(encoded_data: dict):
+ """
+ Decode a previously serialized deadline alert.
+
+ :meta private:
+ """
+ from datetime import timedelta
+
+ from airflow.sdk.serde import deserialize
+ from airflow.serialization.definitions.deadline import SerializedDeadlineAlert
Review Comment:
Handled better in 24548f2e96
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org