External task sensor execution_date_fn

user_11902443 0
Viewed 4915

I want to use execution_date_fn in External task sensor in my dag to make it dependent on previous instance(I do not want to use depends_on_past parameter). Can someone please tell me how can I get previous execution id of the same dag using 'execution_date_fn', so that I do not have to specify hours/minutes in 'execution_delta'. Not sure if below method 'prev_execution_date_1' is correct. Appreciate any help on this.

def prev_execution_date_1(**kwargs):
    dr = self.get_dagrun(session=session)
    previous_scheduled_date = dr.previous_schedule(self.execution_date)
    return previous_scheduled_date


    external_0 = ExternalTaskSensor(
            task_id='Check_Previous_Instance',
            external_task_id=None,
            external_dag_id='dag_abc_1',
            allowed_states=['success'],
            execution_date_fn=prev_execution_date_1,
            dag=dag
    )
airflow sensors directed-acyclic-graphs task dependencies
2 Answers

I got the answer, thought of posting it here, it may help someone.

Below method will return execution id based on if condition I mentioned. In this method, you can see it will return before 2 days execution id if it satisfies the condition, else will return yesterday's execution id.

def prev_execution_dt(execution_date, **kwargs):
    weekday=execution_date.strftime('%A')
    print(weekday)
    if weekday == "Thursday":
        execution_dt_derived=execution_date - timedelta(hours=72)
        print(execution_dt_derived)
    else:
        execution_dt_derived=execution_date - timedelta(hours=24)
        print(execution_dt_derived)
    return execution_dt_derived
user_119024430

execution_date_fn is used to calculate desired execution date according to current execution date if execution_delta is not passed, in current stable version 1.10, there is param check and it accept at most 2 args, context['execution_date'] and context. And context is not include a session, so you could not query database in it. your might try from airflow.settings import Session, but it is not a good idea.

an example is:

lambda dt: dt + timedelta(days=1)

if your scheduler interval is not changed frequently, use execution_delta is a relatively easier way.

user_80250860
Related