ephraimbuddy commented on code in PR #72721:
URL: https://github.com/apache/airflow/pull/72721#discussion_r4006485861
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1511,30 +1517,66 @@ def notify_dagrun_state_changed(self, msg: str):
# we can't get all the state changes on SchedulerJob,
# or LocalTaskJob, so we don't want to "falsely advertise" we notify
about that
+ def _build_callback_last_ti(self, relevant_ti: TI, *, session: Session) ->
TIDataModel | None:
+ """
+ Build a callback context's ``last_ti``, standing in a Dag version if
the record has none.
+
+ ``session`` is required, not defaulted: ``settings.Session`` is a
``scoped_session``,
+ so acquiring one here would hand back the caller's own and then commit
and close it.
+ """
+ from airflow.api_fastapi.execution_api.datamodels.taskinstance import
TaskInstance as TIDataModel
+ from airflow.models.dag_version import DagVersion
+
+ if relevant_ti.dag_version_id is not None:
+ return TIDataModel.model_validate(relevant_ti,
from_attributes=True)
+
+ dag_version_id = self.created_dag_version_id
+ if dag_version_id is None:
+ latest_dag_version = DagVersion.get_latest_version(self.dag_id,
session=session)
+ dag_version_id = latest_dag_version.id if latest_dag_version else
None
+ if dag_version_id is None:
+ self.log.warning(
+ "Task instance %s has no dag_version_id and Dag %s has no
version to stand in; "
+ "omitting last_ti from the Dag callback context.",
+ relevant_ti,
+ self.dag_id,
+ )
+ return None
+ self.log.warning(
Review Comment:
Split the messages so only the latest-version fallback describes
pre-versioning records. Successful fallbacks now log at info level.
--
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]
