From ed2f5d825eceb08e018e3163a47a96221c570f0c Mon Sep 17 00:00:00 2001 From: Ephraim Anierobi Date: Wed, 24 May 2023 10:33:30 +0100 Subject: [PATCH] Remove dependency already registered for this task warning The frequent warning about dependency being already registered for a dag is not needed since task.downstream_task_ids is a set and task.upstream_task_ids is also a set. The warning usually occurs if your tasks are set up similar to this: ``` task1 = mytask() for i in range(5): task2 = anothertask() task1 >> task2 ``` in which case it warns 5 times about task1 dependency is already set. This PR removes the warning --- airflow/models/taskmixin.py | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/airflow/models/taskmixin.py b/airflow/models/taskmixin.py index f0af40e88be70..a858ce942fbaa 100644 --- a/airflow/models/taskmixin.py +++ b/airflow/models/taskmixin.py @@ -208,25 +208,18 @@ def _set_relatives( # If this task does not yet have a dag, add it to the same dag as the other task. self.dag = dag - def add_only_new(obj, item_set: set[str], item: str) -> None: - """Adds only new items to item set.""" - if item in item_set: - self.log.warning("Dependency %s, %s already registered for DAG: %s", obj, item, dag.dag_id) - else: - item_set.add(item) - for task in task_list: if dag and not task.has_dag(): # If the other task does not yet have a dag, add it to the same dag as this task and dag.add_task(task) if upstream: - add_only_new(task, task.downstream_task_ids, self.node_id) - add_only_new(self, self.upstream_task_ids, task.node_id) + task.downstream_task_ids.add(self.node_id) + self.upstream_task_ids.add(task.node_id) if edge_modifier: edge_modifier.add_edge_info(self.dag, task.node_id, self.node_id) else: - add_only_new(self, self.downstream_task_ids, task.node_id) - add_only_new(task, task.upstream_task_ids, self.node_id) + self.downstream_task_ids.add(task.node_id) + task.upstream_task_ids.add(self.node_id) if edge_modifier: edge_modifier.add_edge_info(self.dag, self.node_id, task.node_id)