获取所有Airflow叶节点/任务

Question

我想要构建一个项目，在其中需要捕获所有叶子任务，并向它们添加下游依赖项，以使作业在我们的数据库中完成。在Airflow中是否有一种简单的方法来查找DAG的所有叶子节点？

- Ace Haidrey

---

我现在知道的唯一方法是检查downstream_list并确保它为空。有更好的方法吗？ - Ace Haidrey

Answer 1

使用BaseOperator中的@propertyupstream_task_ids和downstream_task_ids

def get_start_tasks(dag: DAG) -> List[BaseOperator]:
    # returns list of "head" / "root" tasks of DAG
    return [task for task in dag.tasks if not task.upstream_task_ids]


def get_end_tasks(dag: DAG) -> List[BaseOperator]:
    # returns list of "leaf" tasks of DAG
    return [task for task in dag.tasks if not task.downstream_task_ids]

Python 3.6+的Type-Annotations（类型注解）

---

更新-1

现在，Airflow DAG 模型具有强大的@property 函数，例如：

• leaves
• roots
• topological_sort

- y2k-shubham

---

另请参见：获取所有父任务ID - y2k-shubham