DagRun.get_previous_dagrun() was removed

[cristian-fatu]

Apache Airflow version

3.0.3

If "Other Airflow 2 version" selected, which one?

No response

What happened?

Previously, in 2.9.2, I used the dag_run available in the task context and called get_previous_dagrun on it, but from what I understand in Airflow 3 the context holds a stripped down version of the dag_run, which no longer has the get_previous_dagrun method.
Instead, in Airflow 3.0.2 the call fails with:
AttributeError: 'DagRun' object has no attribute 'get_previous_dagrun'

What you think should happen instead?

I think there should be a get_previous_dagrun method as before. The reason I used this in 2.9.2 was to get the state of the previous DagRun and based on the current and previous state decided whether to send a notification or not (e.g.: if a DAG fails, we might only care to get a notification on the first failure or when it recovers, not in between)

How to reproduce

For reference, the code that works in 2.9.2 looks like this:
crt_dag_run = context["dag_run"] previous_dag_run = crt_dag_run.get_previous_dagrun(dag_run=crt_dag_run)

In v3, the DagRun that I get is of type airflow.sdk.api.datamodels._generated.DagRun which doesn't contain the get_previous_dagrun method.

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes 10.6.1
apache-airflow-providers-standard 1.4.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[nrnavaneet]

Would like to work on this :)

[potiuk]

Would like to work on this :)

I am not sure if this is "easy" or whether we want to do much with it. There is a number of methods that have been deliberately removed from "context" objects because they were never intended to be a "public interface" - they were used by people "accidentally" but when we finalized spit of the workers from database - direct DB access (that was needed by this method) is gone and our general recommendation is to use REST API (which was intended for such usage since the beginning of Airlfow 2 - so for about 4 years now).

So I am quite sure we are not going to "fix it" in the sense that this method is not coming back.

I think however that we might think to make some kind of easier replacements for some "common" methods that people used to use - there are a few methods (including get_previous_dag_run) that keeps on repeating and for me this is a signal that we might need to do something about it - either to have such methods in Task SDK directly or making it easier to use REST API from within the tasks.

I will mark it as "affects 3.0" and likely we should discuss it @ashb @kaxil @amoghrajesh -> tagging those who worked most on the task-sdk.

[potiuk]

Ah .. I see @ashb already marked it as "3.0.4" - likely with similar ideas

[nrnavaneet]

Ah alright , got it ! I can migrate the logic to use the REST API instead of relying on get_previous_dagrun(). What do u say? If there are any specifics lmk :). Ill try my best to incorporate them

[potiuk]

Ah alright , got it ! I can migrate the logic to use the REST API instead of relying on get_previous_dagrun(). What do u say? If there are any specifics lmk :). Ill try my best to incorporate them

Well. it requires the user to change it - nothing in Airflow code can be done to change it. User neds to change their code in the Dag currently to use REST API.

[nrnavaneet]

Oh learnt smntg new :)

[kaxil]

The reason I used this in 2.9.2 was to get the state of the previous DagRun and based on the current and previous state decided whether to send a notification or not (e.g.: if a DAG fails, we might only care to get a notification on the first failure or when it recovers, not in between)

@cristian-fatu There is a context["ti"].get_dagrun_state() state method already available -- which you could use in AF 3.0

airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py

Lines 519 to 526 in 3dc4cd2

	def get_dagrun_state(dag_id: str, run_id: str) -> str:
	"""Return the state of the DAG run with the given Run ID."""
	response = SUPERVISOR_COMMS.send(msg=GetDagRunState(dag_id=dag_id, run_id=run_id))
	if TYPE_CHECKING:
	assert isinstance(response, DagRunStateResult)
	return response.state

and

airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py

Lines 496 to 516 in 3dc4cd2

	@staticmethod
	def get_dr_count(
	dag_id: str,
	logical_dates: list[datetime] | None = None,
	run_ids: list[str] | None = None,
	states: list[str] | None = None,
	) -> int:
	"""Return the number of DAG runs matching the given criteria."""
	response = SUPERVISOR_COMMS.send(
	GetDRCount(
	dag_id=dag_id,
	logical_dates=logical_dates,
	run_ids=run_ids,
	states=states,
	),
	)
	if TYPE_CHECKING:
	assert isinstance(response, DRCount)
	return response.count

[cristian-fatu]

@kaxil that would still leave the issue of getting the id of the previous run

[kaxil]

@kaxil that would still leave the issue of getting the id of the previous run

That is true, it is currently generic. We will add something more convenient.