ramitkataria commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3338512929
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -215,46 +216,32 @@ def prune_deadlines(cls, *, session: Session, conditions:
dict[Mapped, Any]) ->
def handle_miss(self, session: Session):
"""Handle a missed deadline by queueing the callback."""
-
- def get_simple_context():
- from airflow.api_fastapi.core_api.datamodels.dag_run import
DAGRunResponse
- from airflow.models import DagRun
-
- # TODO: Use the TaskAPI from within Triggerer to fetch full
context instead of sending this context
- # from the scheduler
-
- # Fetch the DagRun from the database again to avoid errors when
self.dagrun's relationship fields
- # are not in the current session.
- dagrun = session.get(DagRun, self.dagrun_id)
-
- return {
- "dag_run":
DAGRunResponse.model_validate(dagrun).model_dump(mode="json"),
- "deadline": {"id": self.id, "deadline_time":
self.deadline_time},
- }
+ # Routing identifiers: stored at the top level of callback.data so the
triggerer/executor
+ # can locate the DagRun and build execution context *before* invoking
the callback.
+ # These are NOT user payload — they are consumed by infrastructure
(_fetch_callback_dag_run_data
+ # in triggerer_job_runner and _fetch_and_build_context in
callback_supervisor) and stripped
+ # before the callback function is called.
+ # TODO: A dedicated `routing_data` field on Callback would make this
separation explicit
+ # and avoid mixing routing metadata with user kwargs in the same JSON
blob. Deferred because
+ # changing the Callback model would require a migration and
coordination with the executor path.
+ self.callback.data["dag_id"] = self.dagrun.dag_id
+ self.callback.data["run_id"] = self.dagrun.run_id
+
+ # Deadline-specific info goes into kwargs so it's passed to the
callback function.
+ self.callback.data["kwargs"]["deadline_id"] = str(self.id)
Review Comment:
Agreed on moving these out of user `kwargs`. That fixes the `TypeError` for
class-based callables. But I'd like to take it one step further so the two
callback paths stay identical, which is the whole goal here.
Right now only the executor path reconstructs `context["deadline"]`
(`callback_supervisor.py:244-245`). The triggerer path goes through
`build_context_from_dag_run`, which never adds a `deadline` key, so an
`AsyncCallback` has no way to read deadline info at all, and `{{
deadline.deadline_time }}` renders empty.
What if we add `deadline: NotRequired[...]` to the `Context` TypedDict and
build the `deadline` entry **inside the shared `build_context_from_dag_run`
helper** (pass `deadline_id`/`deadline_time` into it). That keeps it a
`Context` (discoverable, in `KNOWN_CONTEXT_KEYS`, type-checked) and lets us
delete the executor-only special case. Both paths then produce identical
context by construction
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
