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
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
...