Programmatic way to stop a particular DAG Run

[nikhil1699]

We have multiple dags running on a kubernetes pod, And we would like to know if it is possible to Stop one of the running DAG with a dag_id and run_id.
There are some ways to get around this on UI but we want a programmatic or CLI way to achieve the same.

https://stackoverflow.com/questions/49039386/how-do-i-stop-an-airflow-dag/54205945

[turbaszek]

The link does not work, this seems to be correct one:
https://stackoverflow.com/questions/49039386/how-do-i-stop-an-airflow-dag

[nikhil1699]

The link does not work, this seems to be correct one:
https://stackoverflow.com/questions/49039386/how-do-i-stop-an-airflow-dag
Yes sorry for inconvenience, But that link couldn't solve our problem. As it shows a UI way.

[turbaszek]

@nikhil1699 I think you may try "stopping" the running DagRun by deleting it or setting it's state to failed. So basically I would suggest checking the code behind the "UI way" and doing the same from script 😉 Or use API:
https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/delete_dag_run

Find a DagRun:

airflow/airflow/models/dagrun.py

Lines 233 to 235 in 4c35955

	def find(
	dag_id: Optional[Union[str, List[str]]] = None,
	run_id: Optional[str] = None,

Change it state:

airflow/airflow/models/dagrun.py

Line 152 in 4c35955

def set_state(self, state):

This won't stop running task as that would require sending a SIGTERM to executor processes.

[turbaszek]

@nikhil1699 did it answer your question?

[nikhil1699]

@turbaszek Thanks for the help. What I am observing was that when I am setting the state to failed, using set_state, it does so in object level, when i use print to check state before and after set_state, it changes the state there, But the DAG doesn't actually stop run.
I think i will have to commit changes to DB as well.
But yes this does solve our problem.

[turbaszek]

I think i will have to commit changes to DB as well.

That's right. You may use this context manager to create session:

airflow/airflow/utils/session.py

Lines 27 to 28 in 4c35955

	def create_session():
	"""Contextmanager that will create and teardown a session."""