pythonairflowjinja2airflow-2.x

Replacement for prev_execution_date as it is deprecated in airflow 2


I am migrating airflow 1 pipelines on airflow 2 and stumbled across the deprecated {{ prev_execution_date }}.

I am not sure what can be used instead.
I found prev_data_interval_start_success and prev_data_interval_end_success but not sure which one to pick from these or are they even correct replacement for prev_execution_date?

There were other template variables like execution_date whose replacement was logical_date.

Doc link : https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html


Solution

  • Just use TaskInstance or DagRun model. Here is an example:

    import logging
    from datetime import datetime
    
    from airflow import DAG
    from airflow.models import TaskInstance, DagRun
    from airflow.operators.python import PythonOperator
    
    dag = DAG(
        dag_id='test_dag',
        start_date=datetime(2024, 1, 1),
        schedule_interval='@daily',
        max_active_runs=1,
    )
    
    
    def first(ti: TaskInstance, **kwargs: dict):
        if ti.previous_ti:
            logging.info('ti.previous_ti.execution_date = %s', ti.previous_ti.execution_date)
        else:
            logging.info('previous_ti not found')
    
        dag_run = DagRun.get_previous_dagrun(dag_run=ti.dag_run)
        if dag_run:
            logging.info('dag_run = %s', dag_run.execution_date)
        else:
            logging.info('dag_run not found')
    
    
    PythonOperator(dag=dag, task_id='first', python_callable=first)
    

    Let's check logs:

    # first dag_run
    ...
    [2024-02-22, 09:44:39 UTC] {test_dag.py:20} INFO - previous_ti not found
    [2024-02-22, 09:44:39 UTC] {test_dag.py:26} INFO - dag_run not found
    ...
    # second dag_run
    [2024-02-22, 09:44:41 UTC] {test_dag.py:18} INFO - ti.previous_ti.execution_date = 2024-01-01 00:00:00+00:00
    [2024-02-22, 09:44:41 UTC] {test_dag.py:24} INFO - dag_run = 2024-01-01 00:00:00+00:00
    ...