From a9a5cc1fca675ec6c375094b0612738291d70f08 Mon Sep 17 00:00:00 2001 From: rohankokkulabito Date: Tue, 23 Jul 2024 23:06:55 +0530 Subject: [PATCH] Update prev_dagrun_dep.py --- airflow/ti_deps/deps/prev_dagrun_dep.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/airflow/ti_deps/deps/prev_dagrun_dep.py b/airflow/ti_deps/deps/prev_dagrun_dep.py index 680377de46575..80545270fc9fe 100644 --- a/airflow/ti_deps/deps/prev_dagrun_dep.py +++ b/airflow/ti_deps/deps/prev_dagrun_dep.py @@ -26,7 +26,7 @@ from airflow.ti_deps.deps.base_ti_dep import BaseTIDep from airflow.utils.db import exists_query from airflow.utils.session import provide_session -from airflow.utils.state import TaskInstanceState +from airflow.utils.state import DagRunState, TaskInstanceState if TYPE_CHECKING: from sqlalchemy.orm import Session @@ -157,7 +157,11 @@ def _get_dep_statuses(self, ti: TI, session: Session, dep_context): return # There was a DAG run, but the task wasn't active back then. - if catchup and last_dagrun.execution_date < ti.task.start_date: + if ( + catchup + and last_dagrun.state != DagRunState.RUNNING + and last_dagrun.execution_date < ti.task.start_date + ): self._push_past_deps_met_xcom_if_needed(ti, dep_context) yield self._passing_status(reason="This task instance was the first task instance for its task.") return