Airflow获取DAG上次运行耗时报错：DagRun无get_last_dagrun属性

问题背景 #

需求为打印上次调度的DAG运行耗时：每日20:00运行的DAG，若上次运行成功则输出耗时，失败则提示失败。编写代码后触发错误：

DagRun has no attribute 'get_last_dagrun'

原代码如下：

from airflow import DAG
from airflow.models import DagRun
from airflow.operators.python_operator import PythonOperator

def print_time_taken():
    dag_run = DagRun.get_last_dagrun(dag_id='my_dag')
    if dag_run and dag_run.state == 'success':
        duration = (dag_run.end_date - dag_run.start_date).total_seconds() / 60
        print(f"The latest DAG run took {duration} minutes to complete.")
    else:
        print("The latest DAG run failed.")

错误原因 #

Airflow的DagRun类并未提供get_last_dagrun这个类方法，该方法不存在导致报错。正确获取最近DAG运行记录的方式是使用DagRun.find()配合排序参数实现。

修正后的代码 #

from airflow import DAG
from airflow.models import DagRun
from airflow.operators.python_operator import PythonOperator
from sqlalchemy import desc

def print_time_taken():
    # 按执行日期倒序查询指定DAG的运行记录，取第一条即为最近一次运行
    dag_runs = DagRun.find(dag_id='my_dag', order_by=[desc(DagRun.execution_date)])
    if dag_runs:
        dag_run = dag_runs[0]
        if dag_run.state == 'success':
            # 计算耗时并保留两位小数
            duration = (dag_run.end_date - dag_run.start_date).total_seconds() / 60
            print(f"The latest DAG run took {duration:.2f} minutes to complete.")
        else:
            # 输出具体的运行状态，便于排查问题
            print(f"The latest DAG run failed with state: {dag_run.state}.")
    else:
        print("No DAG runs found for 'my_dag'.")

代码说明 #

• 使用DagRun.find()方法查询指定DAG的运行记录，通过order_by=[desc(DagRun.execution_date)]按执行日期倒序排列，确保第一条是最近的运行记录
• 增加了空记录判断，避免空列表索引报错
• 优化失败提示，输出具体状态（如failed、up_for_retry等）
• 对耗时结果保留两位小数，提升输出可读性

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