fix(models/dag): handle pre AIP-39 DagRuns

[Lee-W]

Why

Closes: #56451

Back in 3.0.6, we use to have LazyDeserializedDAG.get_run_data_interval

airflow/airflow-core/src/airflow/serialization/serialized_objects.py

Lines 2185 to 2194 in e965c2e

	def get_run_data_interval(self, run: DagRun) -> DataInterval | None:
	"""Get the data interval of this run."""
	if run.dag_id is not None and run.dag_id != self.dag_id:
	raise ValueError(f"Arguments refer to different DAGs: {self.dag_id} != {run.dag_id}")
	data_interval = _get_model_data_interval(run, "data_interval_start", "data_interval_end")
	if data_interval is None and run.logical_date is not None:
	data_interval = self._real_dag.timetable.infer_manual_data_interval(run_after=run.logical_date)
	return data_interval

and DAG.get_run_data_interval

airflow/airflow-core/src/airflow/models/dag.py

Lines 507 to 527 in e965c2e

	def get_run_data_interval(self, run: DagRun) -> DataInterval:
	"""
	Get the data interval of this run.
	For compatibility, this method infers the data interval from the DAG's
	schedule if the run does not have an explicit one set, which is possible for
	runs created prior to AIP-39.
	This function is private to Airflow core and should not be depended on as a
	part of the Python API.
	:meta private:
	"""
	if run.dag_id is not None and run.dag_id != self.dag_id:
	raise ValueError(f"Arguments refer to different DAGs: {self.dag_id} != {run.dag_id}")
	data_interval = _get_model_data_interval(run, "data_interval_start", "data_interval_end")
	if data_interval is not None:
	return data_interval
	# Compatibility: runs created before AIP-39 implementation don't have an
	# explicit data interval. Try to infer from the logical date.
	return self.infer_automated_data_interval(run.logical_date)

They called when we collect Dags.

airflow/airflow-core/src/airflow/dag_processing/collection.py

Line 490 in e965c2e

last_automated_data_interval = dag.get_run_data_interval(last_automated_run)

The dag here is typed as MaybeSerializedDAG which is

airflow/airflow-core/src/airflow/serialization/serialized_objects.py

Line 2081 in e965c2e

MaybeSerializedDAG = Union[DAG, "LazyDeserializedDAG"]

Thus, both LazyDeserializedDAG.get_run_data_interval and DAG.get_run_data_interval will be called back in 3.0.6

However, in 3.1.0, we removed the MaybeSerializedDAG in https://github.com/apache/airflow/pull/54383/files#diff-2f03f28d3201c630dfbae758caf95ada61ab10328de789fc464f6e27c4afb2d0L2687 which make

airflow/airflow-core/src/airflow/dag_processing/collection.py

Line 530 in 54bd5d8

last_automated_data_interval = get_run_data_interval(dag.timetable, last_automated_run)

runs only the logic in DAG.get_run_data_interval.

airflow/airflow-core/src/airflow/models/dag.py

Lines 125 to 143 in 54bd5d8

	def get_run_data_interval(timetable: Timetable, run: DagRun) -> DataInterval:
	"""
	Get the data interval of this run.
	For compatibility, this method infers the data interval from the DAG's
	schedule if the run does not have an explicit one set, which is possible for
	runs created prior to AIP-39.
	This function is private to Airflow core and should not be depended on as a
	part of the Python API.
	:meta private:
	"""
	data_interval = _get_model_data_interval(run, "data_interval_start", "data_interval_end")
	if data_interval is not None:
	return data_interval
	# Compatibility: runs created before AIP-39 implementation don't have an
	# explicit data interval. Try to infer from the logical date.
	return infer_automated_data_interval(timetable, run.logical_date)

What

Since we now only have get_run_data-interval, we should consider both LazyDeserializedDAG.get_run_data_interval and DAG.get_run_data_interval logic

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[Lee-W]

will add some test tomorrow but would appreciate some early review

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test