如何无论DAG触发方式如何都能获取Airflow上一次执行日期？

手动触发Airflow DAG时，内置模板变量prev_execution_date与execution_date取值完全相同，仅在调度器自动触发DAG时二者取值存在正常差异，需要实现无论手动触发还是自动调度触发，都能获取到真实的上一次DAG执行日期。

复现测试代码如下：

echo_exec_date = BashOperator(
    task_id='bash_script',
    bash_command='echo "prev_exec_date={{ prev_execution_date }} execution_date={{ execution_date }}"',
    dag=dag)

手动触发时的运行输出：

prev_exec_date=2022-06-29T08:50:37.506898+00:00 execution_date=2022-06-29T08:50:37.506898+00:00

Airflow默认提供的prev_execution_date模板变量仅针对调度触发的周期运行计算上一周期时间，手动触发DAG时不存在预定义的上一调度周期，变量会直接返回当前运行的execution_date作为值，不会回溯查询实际的历史运行记录。

直接在模板中通过DAG实例的内置方法查询历史运行记录，不受触发方式限制：

方案1：获取最近一次任意状态的历史执行日期 #

包含手动触发、自动触发的所有历史运行，自动排除当前正在运行的实例：

echo_exec_date = BashOperator(
    task_id='bash_script',
    bash_command='''
    {% set last_dagrun = dag.get_last_dagrun(include_externally_triggered=True) %}
    {% set real_prev_exec_date = last_dagrun.execution_date if last_dagrun else None %}
    echo "real_prev_exec_date={{ real_prev_exec_date }} execution_date={{ execution_date }}"
    ''',
    dag=dag
)

参数说明：

• include_externally_triggered=True：将手动触发的历史运行纳入统计，设为False则仅统计调度器自动触发的运行
• DAG首次运行无历史记录时，real_prev_exec_date返回None，可根据业务需要替换为自定义默认值

方案2：获取最近一次成功状态的历史执行日期 #

如果需要过滤失败、运行中状态的实例，仅取最近一次运行成功的执行日期，可使用如下写法：

echo_exec_date = BashOperator(
    task_id='bash_script',
    bash_command='''
    {% set success_runs = dag.get_dagruns_between(
        start_date=dag.default_args['start_date'],
        end_date=execution_date,
        state='success'
    ) %}
    {% set last_success_run = success_runs[-1] if success_runs else None %}
    {% set real_prev_exec_date = last_success_run.execution_date if last_success_run else None %}
    echo "real_prev_exec_date={{ real_prev_exec_date }} execution_date={{ execution_date }}"
    ''',
    dag=dag
)

内容的提问来源于stack exchange，提问作者kms333